1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
# This file is part of Senator Sperma
# Licensed under the GPL-3.0 License
# Copyright (C) 2026 Salmonik @ salmonik@tuta.io
# https://codeberg.org/salmonik/senator-sperma
import os
import aiofiles
import requests
import urllib.parse
from typing import List, Optional
import discord
from discord.ext import commands
from discord import AutocompleteContext
from cogs.utils import file_search as search_utils
import logging
logger = logging.getLogger(__name__)
class Uploader(commands.Cog):
def __init__(self, bot):
from cogs.utils.temp import get_temp_dir
self.bot = bot
self.temp_dir = get_temp_dir()
self._cached_urls = {}
self._uploadurl = os.getenv("COPYPARTY_URL_UPLOAD")
async def _file_autocomplete(self, ctx: AutocompleteContext):
query = ctx.value.lower() if ctx.value else ""
results = await search_utils.UniversalSearcher.search_copyparty_autocomplete(query, max_results=25)
user_id = ctx.interaction.user.id
self._cached_urls[user_id] = {urllib.parse.unquote(url.split('/')[-1]): url for url in results}
choices = []
for url in results[:25]:
filename = urllib.parse.unquote(url.split('/')[-1])
display_name = filename[:99] + "…" if len(filename) > 100 else filename
choices.append(discord.OptionChoice(name=display_name, value=filename))
return choices
async def _download_from_url(self, url: str) -> Optional[str]:
"""Download file from copyparty URL and return local path"""
try:
response = requests.get(url)
if response.status_code != 200:
return None
filename = url.split('/')[-1].split('?')[0]
file_path = os.path.join(self.temp_dir, filename)
async with aiofiles.open(file_path, 'wb') as f:
await f.write(response.content)
return file_path
except Exception as e:
logger.error(f"Error downloading from {url}: {e}")
return None
async def _upload_to_copyparty(self, file_path: str) -> Optional[str]:
"""Upload file to copyparty and return download URL"""
try:
async with aiofiles.open(file_path, 'rb') as f:
file_content = await f.read()
files = {'f': (os.path.basename(file_path), file_content, 'application/octet-stream')}
headers = {'Cookie': f'cppwd={os.getenv("COPYPARTY_PASSWORD")}'}
params = {'j': '1', 'ck': 'b2'}
response = requests.post(self._uploadurl, files=files, headers=headers, params=params)
if response.status_code in [200, 201]:
result = response.json()
return result.get('fileurl')
except Exception as e:
logger.error(f"Upload error: {e}")
return None
async def _upload_file(self, file_path: str) -> Optional[str]:
"""Upload a file and return the URL"""
if not self._uploadurl:
logger.error("COPYPARTY_URL_UPLOAD not set!")
return None
return await self._upload_to_copyparty(file_path)
@commands.message_command(
name="upload",
description="wyślij załączniki na copyparty",
)
async def message_upload(self, ctx, message: discord.Message):
if not message.attachments:
await ctx.respond("Brak załączników w wiadomości", ephemeral=True)
return
await ctx.defer()
uploaded_urls = []
failed_count = 0
for attachment in message.attachments:
file_path = os.path.join(self.temp_dir, attachment.filename)
await attachment.save(file_path)
url = await self._upload_file(file_path)
if url:
uploaded_urls.append(url)
else:
failed_count += 1
if os.path.exists(file_path):
os.remove(file_path)
if uploaded_urls:
response = f"Wysłano {len(uploaded_urls)} plików:\n" + "\n".join(uploaded_urls)
if failed_count > 0:
response += f"\n\nNie wysłano {failed_count} plików"
await ctx.respond(response)
else:
await ctx.respond("Nie udało się wysłać żadnych plików", ephemeral=True)
@commands.slash_command(
name="upload",
description="wyślij plik na copyparty",
)
@discord.option(
name="zalacznik",
description="plik z discorda",
required=True,
type=discord.Attachment
)
async def upload(self, ctx, zalacznik: discord.Attachment = None):
await ctx.defer()
file_path = None
file_path = os.path.join(self.temp_dir, zalacznik.filename)
await zalacznik.save(file_path)
if not file_path or not os.path.exists(file_path):
await ctx.respond("Nie udało się pobrać pliku", ephemeral=True)
return
try:
dl_url = await self._upload_file(file_path)
if dl_url:
await ctx.respond(dl_url)
else:
await ctx.respond("Nie udało się wysłać pliku", ephemeral=True)
finally:
if file_path and os.path.exists(file_path):
os.remove(file_path)
def setup(bot):
bot.add_cog(Uploader(bot))
|
