# FILE: _St/Analyze8780/analyze8780_main.py | ROLE: Analyze8780 HTTP API·정적 UI·로컬 미디어 서비스 from __future__ import annotations import argparse import json import mimetypes import re from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, quote, urlparse from analyze8780_repository import AnalyzeRepository VERSION = "2.3.8.297" DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 8780 class AnalyzeHandler(BaseHTTPRequestHandler): repository: AnalyzeRepository web_root: Path def log_message(self, fmt: str, *args: object) -> None: print(f"[Analyze8780] {self.address_string()} {fmt % args}") def _cors(self) -> None: self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET,PUT,POST,DELETE,OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type,Range") self.send_header("Cache-Control", "no-store") def _json(self, data: object, status: int = 200) -> None: raw = json.dumps(data, ensure_ascii=False, separators=(",", ":")).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self._cors() self.end_headers() self.wfile.write(raw) def _body_json(self) -> dict: length = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(length) if length else b"{}" value = json.loads(raw.decode("utf-8")) if not isinstance(value, dict): raise ValueError("JSON object required") return value def _static(self, relative: str) -> None: rel = relative.strip("/") or "index.html" target = (self.web_root / rel).resolve() if self.web_root not in target.parents and target != self.web_root: self._json({"ok": False, "error": "invalid_path"}, 400) return if not target.is_file(): self._json({"ok": False, "error": "not_found"}, 404) return raw = target.read_bytes() content_type = mimetypes.guess_type(target.name)[0] or "application/octet-stream" self.send_response(200) self.send_header("Content-Type", content_type + ("; charset=utf-8" if content_type.startswith("text/") else "")) self.send_header("Content-Length", str(len(raw))) self._cors() self.end_headers() self.wfile.write(raw) def _media(self, target: Path, filename: str) -> None: size = target.stat().st_size start = 0 end = max(0, size - 1) status = HTTPStatus.OK range_header = str(self.headers.get("Range") or "").strip() if range_header: match = re.fullmatch(r"bytes=(\d*)-(\d*)", range_header) if not match: self.send_response(HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE) self.send_header("Content-Range", f"bytes */{size}") self._cors() self.end_headers() return left, right = match.groups() if left: start = int(left) end = int(right) if right else end elif right: length = min(size, int(right)) start = max(0, size - length) if start >= size or end < start: self.send_response(HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE) self.send_header("Content-Range", f"bytes */{size}") self._cors() self.end_headers() return end = min(end, size - 1) status = HTTPStatus.PARTIAL_CONTENT length = max(0, end - start + 1) content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream" self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(length)) self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Disposition", f"inline; filename*=UTF-8''{quote(filename)}") if status == HTTPStatus.PARTIAL_CONTENT: self.send_header("Content-Range", f"bytes {start}-{end}/{size}") self._cors() self.end_headers() with target.open("rb") as handle: handle.seek(start) remaining = length while remaining > 0: chunk = handle.read(min(1024 * 1024, remaining)) if not chunk: break self.wfile.write(chunk) remaining -= len(chunk) @staticmethod def _int(params: dict[str, list[str]], key: str, default: int) -> int: try: return int((params.get(key) or [default])[0]) except (TypeError, ValueError): return default def do_OPTIONS(self) -> None: self.send_response(204) self._cors() self.end_headers() def do_GET(self) -> None: parsed = urlparse(self.path) path = parsed.path params = parse_qs(parsed.query, keep_blank_values=True) try: if path == "/api/health": self._json(self.repository.health()) return if path == "/api/accounts": self._json(self.repository.accounts( page=self._int(params, "page", 1), page_size=self._int(params, "page_size", 50), include_deleted=(params.get("include_deleted") or ["0"])[0] == "1", search_pattern=(params.get("search") or [""])[0], )) return if path == "/api/accounts/check": self._json(self.repository.account_table_check((params.get("search") or [""])[0])) return if path == "/api/collection-accounts": self._json(self.repository.collection_accounts()) return if path == "/api/database-files": self._json(self.repository.database_files()) return if path == "/api/deleted/accounts": self._json({"ok": True, "items": self.repository.deleted_accounts()}) return if path.startswith("/_StDown/instagram/"): target = self.repository.resolve_stdown_virtual_path(path) self._media(target, target.name) return match = re.fullmatch(r"/api/accounts/(\d+)/posts", path) if match: self._json(self.repository.account_posts( int(match.group(1)), page=self._int(params, "page", 1), page_size=self._int(params, "page_size", 50), sort_by=(params.get("sort_by") or ["views"])[0], )) return match = re.fullmatch(r"/api/posts/(\d+)/stats", path) if match: self._json({"ok": True, "items": self.repository.post_stats(int(match.group(1)), limit=self._int(params, "limit", 144))}) return match = re.fullmatch(r"/api/posts/(\d+)", path) if match: item = self.repository.post(int(match.group(1))) self._json({"ok": bool(item), "item": item}, 200 if item else 404) return if path in {"/", "/index.htm", "/index.html"}: self._static("index.html") return if path.startswith("/static/"): self._static(path[len("/static/"):]) return self._json({"ok": False, "error": "not_found"}, 404) except FileNotFoundError as exc: self._json({"ok": False, "error": str(exc)}, 404) except Exception as exc: self._json({"ok": False, "error": str(exc)}, 400) def do_PUT(self) -> None: parsed = urlparse(self.path) try: if parsed.path == "/api/collection-accounts": self._json(self.repository.replace_collection_accounts(self._body_json())) return self._json({"ok": False, "error": "not_found"}, 404) except Exception as exc: self._json({"ok": False, "error": str(exc)}, 400) def do_POST(self) -> None: parsed = urlparse(self.path) try: if parsed.path == "/api/observations": result = self.repository.save_observation(self._body_json()) self._json({"ok": True, **result}, HTTPStatus.CREATED) return if parsed.path == "/api/observations/batch": result = self.repository.save_observations(self._body_json()) self._json(result, HTTPStatus.CREATED if result.get("ok") else HTTPStatus.MULTI_STATUS) return if parsed.path == "/api/database/open": self._json(self.repository.open_database(self._body_json())) return if parsed.path == "/api/database/new": self._json(self.repository.new_database(self._body_json()), HTTPStatus.CREATED) return if parsed.path == "/api/database/save-as": self._json(self.repository.save_database_as(self._body_json()), HTTPStatus.CREATED) return if parsed.path == "/api/open-folder": payload = self._body_json() self._json({"ok": True, **self.repository.open_local_media_folder(payload.get("virtual_path"))}) return self._json({"ok": False, "error": "not_found"}, 404) except FileNotFoundError as exc: self._json({"ok": False, "error": str(exc)}, 404) except Exception as exc: self._json({"ok": False, "error": str(exc)}, 400) def do_DELETE(self) -> None: path = urlparse(self.path).path try: match = re.fullmatch(r"/api/accounts/(\d+)", path) if match: self._json({"ok": True, **self.repository.delete_account(int(match.group(1)))}) return match = re.fullmatch(r"/api/posts/(\d+)", path) if match: self._json({"ok": True, **self.repository.delete_post(int(match.group(1)))}) return self._json({"ok": False, "error": "not_found"}, 404) except Exception as exc: self._json({"ok": False, "error": str(exc)}, 400) def build_server(st_root: Path, db_path: Path | None, host: str, port: int) -> ThreadingHTTPServer: repo = AnalyzeRepository(st_root, db_path) handler = type("BoundAnalyzeHandler", (AnalyzeHandler,), { "repository": repo, "web_root": Path(__file__).resolve().parent / "web", }) return ThreadingHTTPServer((host, port), handler) def main() -> int: parser = argparse.ArgumentParser(description="sort local analysis service on port 8780") parser.add_argument("--host", default=DEFAULT_HOST) parser.add_argument("--port", type=int, default=DEFAULT_PORT) parser.add_argument("--st-root", default=str(Path(__file__).resolve().parents[1])) parser.add_argument("--db") args = parser.parse_args() server = build_server(Path(args.st_root), Path(args.db) if args.db else None, args.host, args.port) print(json.dumps({"overall": "OK", "service": "Analyze8780", "version": VERSION, "url": f"http://{args.host}:{args.port}/"}, ensure_ascii=False)) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())