# retoor import asyncio import ipaddress import logging import socket from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlparse from PIL import Image from io import BytesIO import httpx from devplacepy import stealth from devplacepy.database import get_table, db, get_setting from devplacepy.config import UPLOADS_DIR, ATTACHMENTS_DIR from devplacepy.utils import generate_uid logger = logging.getLogger(__name__) REMOTE_FETCH_TIMEOUT = 20.0 REMOTE_FETCH_USER_AGENT = ( "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" ) THUMBNAIL_SIZE = (200, 200) THUMBNAIL_QUALITY = 80 IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".tiff"} THUMBNAIL_EXTENSIONS = IMAGE_EXTENSIONS - {".gif"} POST_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp"} ALLOWED_UPLOAD_TYPES = { ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp", ".tiff": "image/tiff", ".pdf": "application/pdf", ".zip": "application/zip", ".mp4": "video/mp4", ".webm": "video/webm", ".ogv": "video/ogg", ".mov": "video/quicktime", ".m4v": "video/x-m4v", ".mp3": "audio/mpeg", ".txt": "text/plain", ".py": "text/x-python", ".js": "text/javascript", ".css": "text/css", ".md": "text/markdown", ".wav": "audio/wav", ".flac": "audio/flac", ".ogg": "audio/ogg", ".aac": "audio/aac", ".wma": "audio/x-ms-wma", ".m4a": "audio/mp4", ".avi": "video/x-msvideo", ".mkv": "video/x-matroska", ".flv": "video/x-flv", ".wmv": "video/x-ms-wmv", ".3gp": "video/3gpp", ".csv": "text/csv", ".doc": "application/msword", ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ".xls": "application/vnd.ms-excel", ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ".ppt": "application/vnd.ms-powerpoint", ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation", ".odt": "application/vnd.oasis.opendocument.text", ".rtf": "application/rtf", ".json": "application/json", ".xml": "application/xml", ".yaml": "text/yaml", ".yml": "text/yaml", ".toml": "text/x-toml", ".sh": "text/x-sh", ".bat": "text/x-bat", ".ts": "text/typescript", ".java": "text/x-java", ".cpp": "text/x-c++", ".c": "text/x-c", ".h": "text/x-c-header", ".rb": "text/x-ruby", ".go": "text/x-go", ".rs": "text/x-rust", ".sql": "text/x-sql", ".php": "text/x-php", ".swift": "text/x-swift", ".kt": "text/x-kotlin", ".cfg": "text/x-config", ".ini": "text/x-config", ".log": "text/plain", ".tar": "application/x-tar", ".gz": "application/gzip", ".rar": "application/vnd.rar", ".7z": "application/x-7z-compressed", } MIME_TO_EXT = { "image/jpeg": ".jpg", "image/png": ".png", "image/gif": ".gif", "image/webp": ".webp", "image/bmp": ".bmp", "image/tiff": ".tiff", "application/pdf": ".pdf", "application/zip": ".zip", "video/mp4": ".mp4", "video/webm": ".webm", "video/ogg": ".ogv", "video/quicktime": ".mov", "video/x-m4v": ".m4v", "audio/mpeg": ".mp3", "text/plain": ".txt", "text/markdown": ".md", "audio/wav": ".wav", "audio/flac": ".flac", "audio/ogg": ".ogg", "audio/aac": ".aac", "audio/x-ms-wma": ".wma", "audio/mp4": ".m4a", "video/x-msvideo": ".avi", "video/x-matroska": ".mkv", "video/x-flv": ".flv", "video/x-ms-wmv": ".wmv", "video/3gpp": ".3gp", "text/csv": ".csv", "application/msword": ".doc", "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx", "application/vnd.ms-excel": ".xls", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx", "application/vnd.ms-powerpoint": ".ppt", "application/vnd.openxmlformats-officedocument.presentationml.presentation": ".pptx", "application/vnd.oasis.opendocument.text": ".odt", "application/rtf": ".rtf", "application/json": ".json", "application/xml": ".xml", "text/yaml": ".yaml", "text/x-toml": ".toml", "text/x-sh": ".sh", "text/x-bat": ".bat", "text/typescript": ".ts", "text/x-java": ".java", "text/x-c++": ".cpp", "text/x-c": ".c", "text/x-c-header": ".h", "text/x-ruby": ".rb", "text/x-go": ".go", "text/x-rust": ".rs", "text/x-sql": ".sql", "text/x-php": ".php", "text/x-swift": ".swift", "text/x-kotlin": ".kt", "text/x-config": ".cfg", "application/x-tar": ".tar", "application/gzip": ".gz", "application/vnd.rar": ".rar", "application/x-7z-compressed": ".7z", } FILE_ICONS = { ".pdf": "\U0001f4c4", ".zip": "\U0001f4e6", ".gz": "\U0001f4e6", ".tar": "\U0001f4e6", ".rar": "\U0001f4e6", ".7z": "\U0001f4e6", ".mp4": "\U0001f3ac", ".webm": "\U0001f3ac", ".ogv": "\U0001f3ac", ".mov": "\U0001f3ac", ".m4v": "\U0001f3ac", ".mp3": "\U0001f3b5", ".py": "\U0001f4bb", ".js": "\U0001f4bb", ".ts": "\U0001f4bb", ".html": "\U0001f4bb", ".css": "\U0001f4bb", ".json": "\U0001f4bb", ".md": "\U0001f4bb", ".csv": "\U0001f4ca", ".xls": "\U0001f4ca", ".xlsx": "\U0001f4ca", ".doc": "\U0001f4dd", ".docx": "\U0001f4dd", ".txt": "\U0001f4c4", ".exe": "\u2699", ".bin": "\u2699", } DEFAULT_FILE_ICON = "\U0001f4ce" def _get_max_upload_bytes(): return int(get_setting("max_upload_size_mb", "10")) * 1024 * 1024 WILDCARD_TOKENS = {"*", ".*", "*.*"} def allowed_extensions(): raw = get_setting("allowed_file_types", "").strip() if not raw: return set(ALLOWED_UPLOAD_TYPES) tokens = {part.strip().lower() for part in raw.split(",") if part.strip()} if tokens & WILDCARD_TOKENS: return set(ALLOWED_UPLOAD_TYPES) return {token if token.startswith(".") else f".{token}" for token in tokens} def is_extension_allowed(ext): return ext in allowed_extensions() def _directory_for(uid): tail = uid.replace("-", "") return f"{tail[-2:]}/{tail[-4:-2]}" def _detect_mime(file_bytes, original_filename): ext = Path(original_filename).suffix.lower() return ALLOWED_UPLOAD_TYPES.get(ext, "application/octet-stream") def _image_dimensions(file_bytes): try: img = Image.open(BytesIO(file_bytes)) return img.width, img.height except Exception as e: logger.warning(f"Could not read image dimensions: {e}") return None, None def _generate_thumbnail(file_bytes, thumb_path): try: img = Image.open(BytesIO(file_bytes)) img.thumbnail(THUMBNAIL_SIZE, Image.LANCZOS) if img.mode in ("RGBA", "P"): img = img.convert("RGBA") thumb_path = thumb_path.with_suffix(".png") img.save(str(thumb_path), "PNG") return thumb_path.name img = img.convert("RGB") img.save(str(thumb_path), "JPEG", quality=THUMBNAIL_QUALITY) return thumb_path.name except Exception as e: logger.warning(f"Thumbnail generation failed: {e}") return None def save_inline_image(file_bytes, original_filename): if len(file_bytes) > _get_max_upload_bytes(): logger.warning(f"Inline image too large: {original_filename}") return None ext = Path(original_filename).suffix.lower() if ext not in POST_IMAGE_EXTENSIONS: logger.warning(f"Unsupported inline image type: {ext}") return None UPLOADS_DIR.mkdir(parents=True, exist_ok=True) filename = f"{generate_uid()}{ext}" (UPLOADS_DIR / filename).write_bytes(file_bytes) logger.info(f"Inline image saved: {filename}") return filename def delete_inline_image(filename): if not filename: return try: (UPLOADS_DIR / filename).unlink(missing_ok=True) except Exception as e: logger.warning(f"Failed to delete inline image {filename}: {e}") def store_attachment(file_bytes, original_filename, user_uid): if len(file_bytes) > _get_max_upload_bytes(): return None ext = Path(original_filename).suffix.lower() if not is_extension_allowed(ext): return None uid = generate_uid() stored_name = f"{uid}{ext}" directory = _directory_for(uid) file_dir = ATTACHMENTS_DIR / directory file_dir.mkdir(parents=True, exist_ok=True) (file_dir / stored_name).write_bytes(file_bytes) mime = _detect_mime(file_bytes, original_filename) is_image = mime.startswith("image/") image_width, image_height = None, None thumbnail = None if is_image: image_width, image_height = _image_dimensions(file_bytes) if ext not in (".gif",): thumbnail = _generate_thumbnail(file_bytes, file_dir / f"{uid}_thumb.jpg") is_audio = mime.startswith("audio/") get_table("attachments").insert( { "uid": uid, "target_type": "", "target_uid": "", "user_uid": user_uid, "original_filename": original_filename, "stored_name": stored_name, "directory": directory, "file_size": len(file_bytes), "mime_type": mime, "image_width": image_width, "image_height": image_height, "has_thumbnail": 1 if thumbnail else 0, "thumbnail_name": thumbnail, "gitea_asset_id": None, "created_at": datetime.now(timezone.utc).isoformat(), "deleted_at": None, "deleted_by": None, } ) return { "uid": uid, "original_filename": original_filename, "file_size": len(file_bytes), "mime_type": mime, "url": f"/static/uploads/attachments/{directory}/{stored_name}", "thumbnail_url": f"/static/uploads/attachments/{directory}/{thumbnail}" if thumbnail else None, "has_thumbnail": thumbnail is not None, "is_image": is_image, "is_video": mime.startswith("video/"), "is_audio": is_audio, } class RemoteFetchError(Exception): def __init__(self, message, status=400): super().__init__(message) self.message = message self.status = status async def _guard_public_url(url): parsed = urlparse(url) if parsed.scheme not in ("http", "https"): raise RemoteFetchError("Only http and https URLs can be attached.", 400) host = parsed.hostname if not host: raise RemoteFetchError("The URL has no host.", 400) try: infos = await asyncio.to_thread(socket.getaddrinfo, host, None) except socket.gaierror as exc: raise RemoteFetchError(f"Could not resolve host: {host}", 400) from exc for info in infos: address = ipaddress.ip_address(info[4][0]) if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped: address = address.ipv4_mapped if ( address.is_private or address.is_loopback or address.is_link_local or address.is_reserved or address.is_multicast or address.is_unspecified ): raise RemoteFetchError( f"Refusing to attach a private or local address ({address}).", 400 ) def _resolve_remote_filename(final_url, content_type, override): name = (override or "").strip() or Path(urlparse(final_url).path).name ext = Path(name).suffix.lower() if name and ext and is_extension_allowed(ext): return name base_mime = (content_type or "").split(";")[0].strip().lower() mapped = MIME_TO_EXT.get(base_mime) if mapped is None or not is_extension_allowed(mapped): return None stem = Path(name).stem or "download" return f"{stem}{mapped}" async def fetch_remote_file(url, filename=None): if "://" not in url: url = "https://" + url await _guard_public_url(url) max_bytes = _get_max_upload_bytes() try: async with stealth.stealth_async_client( follow_redirects=True, timeout=REMOTE_FETCH_TIMEOUT, headers={"User-Agent": REMOTE_FETCH_USER_AGENT}, ) as client: async with client.stream("GET", url) as response: if response.status_code >= 400: raise RemoteFetchError( f"The remote server returned {response.status_code}.", 400 ) final_url = str(response.url) content_type = response.headers.get("content-type", "") chunks = [] total = 0 async for chunk in response.aiter_bytes(): chunks.append(chunk) total += len(chunk) if total > max_bytes: raise RemoteFetchError( f"The file exceeds the {max_bytes // (1024 * 1024)}MB limit.", 413, ) data = b"".join(chunks) except httpx.HTTPError as exc: raise RemoteFetchError(f"Could not fetch {url}: {exc}", 400) from exc name = _resolve_remote_filename(final_url, content_type, filename) if name is None: raise RemoteFetchError( "Could not determine an allowed file type for the URL. Pass a filename " "with an allowed extension.", 415, ) return name, data async def store_attachment_from_url(url, user_uid, filename=None): name, data = await fetch_remote_file(url, filename) result = store_attachment(data, name, user_uid) if result is None: raise RemoteFetchError( "The downloaded file is not an allowed type or exceeds the size limit.", 413, ) return result def link_attachments(uids, target_type, target_uid): flat = [ uid.strip() for raw in uids or [] for uid in str(raw).split(",") if uid.strip() ] if not flat: return placeholders = ",".join(f":p{i}" for i in range(len(flat))) params = {f"p{i}": uid for i, uid in enumerate(flat)} with db: db.query( f"UPDATE attachments SET target_type=:tt, target_uid=:tu WHERE uid IN ({placeholders})", tt=target_type, tu=target_uid, **params, ) def set_gitea_asset_id(uid, asset_id): get_table("attachments").update( {"uid": uid, "gitea_asset_id": int(asset_id)}, ["uid"] ) async def mirror_attachment_to_gitea(uid): from devplacepy.services.gitea import runtime from devplacepy.services.gitea.client import GiteaError row = get_table("attachments").find_one(uid=uid, deleted_at=None) if not row: return None target_type = row.get("target_type", "") target_uid = row.get("target_uid", "") if target_type not in ("issue", "issue_comment") or not target_uid: return None path = ATTACHMENTS_DIR / row.get("directory", "") / row.get("stored_name", "") try: data = path.read_bytes() except OSError as exc: logger.warning("Cannot read attachment %s for Gitea mirror: %s", uid, exc) return None filename = row.get("original_filename") or row.get("stored_name") or "file" mime = row.get("mime_type") or "application/octet-stream" client = runtime.get_client() try: if target_type == "issue": asset = await client.create_issue_asset( int(target_uid), filename, data, mime ) else: asset = await client.create_comment_asset( int(target_uid), filename, data, mime ) except (GiteaError, ValueError) as exc: logger.warning("Gitea asset mirror failed for %s: %s", uid, exc) return None asset_id = int(asset.get("id", 0) or 0) if asset_id: set_gitea_asset_id(uid, asset_id) return asset_id async def remove_gitea_asset(row): from devplacepy.services.gitea import runtime from devplacepy.services.gitea.client import GiteaError asset_id = int(row.get("gitea_asset_id") or 0) target_type = row.get("target_type", "") target_uid = row.get("target_uid", "") if not asset_id or not target_uid: return client = runtime.get_client() try: if target_type == "issue": await client.delete_issue_asset(int(target_uid), asset_id) elif target_type == "issue_comment": await client.delete_comment_asset(int(target_uid), asset_id) except (GiteaError, ValueError) as exc: logger.warning("Gitea asset delete failed for %s: %s", row.get("uid"), exc) def _unlink_attachment_files(row): stored_name = row.get("stored_name", "") directory = row.get("directory", "") if not (stored_name and directory): return file_path = ATTACHMENTS_DIR / directory / stored_name try: file_path.unlink(missing_ok=True) except Exception as e: logger.warning(f"Failed to delete attachment file {file_path}: {e}") for thumb_path in (ATTACHMENTS_DIR / directory).glob( f"{Path(stored_name).stem}_thumb.*" ): try: thumb_path.unlink(missing_ok=True) except Exception as e: logger.warning(f"Failed to delete thumbnail {thumb_path}: {e}") def _delete_attachment_row(row): _unlink_attachment_files(row) get_table("attachments").delete(id=row["id"]) def delete_attachment(uid): row = get_table("attachments").find_one(uid=uid) if row: _delete_attachment_row(row) def rename_attachment(uid, filename): row = get_table("attachments").find_one(uid=uid, deleted_at=None) if not row: return None ext = Path(row.get("stored_name", "")).suffix.lower() stem = Path(str(filename)).name.strip() if ext: stem = Path(stem).stem if not stem: return None clean = f"{stem}{ext}" get_table("attachments").update({"uid": uid, "original_filename": clean}, ["uid"]) return clean def soft_delete_attachment(uid, deleted_by="system"): row = get_table("attachments").find_one(uid=uid) if not row or row.get("deleted_at"): return None get_table("attachments").update( { "uid": uid, "deleted_at": datetime.now(timezone.utc).isoformat(), "deleted_by": deleted_by, }, ["uid"], ) return row def restore_attachment(uid): row = get_table("attachments").find_one(uid=uid) if not row or not row.get("deleted_at"): return False get_table("attachments").update( {"uid": uid, "deleted_at": None, "deleted_by": None}, ["uid"] ) return True def soft_delete_target_attachments(target_type, target_uid, deleted_by): stamp = datetime.now(timezone.utc).isoformat() for row in get_table("attachments").find( target_type=target_type, target_uid=target_uid, deleted_at=None ): get_table("attachments").update( {"uid": row["uid"], "deleted_at": stamp, "deleted_by": deleted_by}, ["uid"] ) def soft_delete_attachments_for(target_type, target_uids, deleted_by): from devplacepy.database import soft_delete_in soft_delete_in( "attachments", "target_uid", target_uids, deleted_by, target_type=target_type, ) def delete_target_attachments(target_type, target_uid): for row in get_table("attachments").find( target_type=target_type, target_uid=target_uid ): _delete_attachment_row(row) def delete_attachments_for(target_type, target_uids): uids = [uid for uid in target_uids if uid] if not uids or "attachments" not in db.tables: return placeholders = ",".join(f":p{i}" for i in range(len(uids))) params = {f"p{i}": uid for i, uid in enumerate(uids)} rows = list( db.query( f"SELECT * FROM attachments WHERE target_type=:tt AND target_uid IN ({placeholders})", tt=target_type, **params, ) ) if not rows: return for row in rows: _unlink_attachment_files(row) ids = ",".join(str(row["id"]) for row in rows) with db: db.query(f"DELETE FROM attachments WHERE id IN ({ids})") def get_attachments(target_type, target_uid): if "attachments" not in db.tables: return [] rows = list( get_table("attachments").find( target_type=target_type, target_uid=target_uid, deleted_at=None, order_by=["created_at"], ) ) return [_row_to_attachment(r) for r in rows] def get_attachments_batch(target_type, uids): if not uids: return {} if "attachments" not in db.tables: return {uid: [] for uid in uids} placeholders = ",".join(f":p{i}" for i in range(len(uids))) params = {f"p{i}": uid for i, uid in enumerate(uids)} rows = db.query( f"SELECT * FROM attachments WHERE target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL ORDER BY created_at", tt=target_type, **params, ) result = {uid: [] for uid in uids} for row in rows: if row["target_uid"] in result: result[row["target_uid"]].append(_row_to_attachment(row)) return result def _row_to_attachment(row): stored_name = row.get("stored_name", "") directory = row.get("directory", "") thumb_name = None if row.get("has_thumbnail"): thumb_name = row.get("thumbnail_name") if not thumb_name: stem = Path(stored_name).stem png = f"{stem}_thumb.png" thumb_name = ( png if (ATTACHMENTS_DIR / directory / png).exists() else f"{stem}_thumb.jpg" ) return { "uid": row["uid"], "original_filename": row.get("original_filename", ""), "file_size": row.get("file_size", 0), "mime_type": row.get("mime_type", ""), "url": f"/static/uploads/attachments/{directory}/{stored_name}", "thumbnail_url": f"/static/uploads/attachments/{directory}/{thumb_name}" if thumb_name else None, "has_thumbnail": bool(row.get("has_thumbnail")), "is_image": row.get("mime_type", "").startswith("image/"), "is_video": row.get("mime_type", "").startswith("video/"), "is_audio": row.get("mime_type", "").startswith("audio/"), "target_type": row.get("target_type", ""), "target_uid": row.get("target_uid", ""), "user_uid": row.get("user_uid", ""), "gitea_asset_id": row.get("gitea_asset_id") or None, "created_at": row.get("created_at", ""), } def format_file_size(size): if size < 1024: return f"{size} B" if size < 1048576: return f"{size / 1024:.1f} KB" return f"{size / 1048576:.1f} MB" def file_icon_emoji(filename): ext = Path(filename).suffix.lower() return FILE_ICONS.get(ext, DEFAULT_FILE_ICON)