# -*- coding: utf-8 -*- # FILE: _St/Cache8701/cache8701_main.py | ROLE: Chrome CDP 캐시·응답 이미지 추출 서비스 from __future__ import annotations import argparse import base64 import json import os import re import socket import struct import threading import time import traceback import urllib.parse import urllib.request import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from datetime import datetime from pathlib import Path from typing import Any, Dict, Optional ROOT = Path(__file__).resolve().parents[1] CONFIG_PATH = Path(__file__).resolve().with_name("cache8701_config.json") STATE_LOCK = threading.RLock() OBSERVED: Dict[str, dict] = {} # key: tab_id|normalized_image_url SAVED: Dict[str, float] = {} CACHE_RUNS: Dict[str, dict] = {} # [Cache Image 완료 기준][회귀 금지] # 자동 MP4 저장 여부와 무관하게 관찰된 이미지 메타데이터를 처리한다. # 파일은 {shortcode}.{실제확장자}, 상태 원본은 SQLite이며 해시·정적 보고서를 새로 만들지 않는다. LOG_ONCE: Dict[str, str] = {} LOG_LOCK = threading.RLock() LOG_FILE = None LOG_FILE_PATH = "" STOP = threading.Event() CONFIG: dict = {} RUNTIME: dict = { "connection": {}, "watcher": None, "watcher_stop": None, "watcher_signature": "", "sessions": {}, "last_error": "", } def load_config() -> dict: base = { "schema": "cache8701_config_v2", "version": "2.3.8.324", "host": "127.0.0.1", "port": 8701, "auto8700_host": "127.0.0.1", "auto8700_port": 8707, "auto8700_connection_path": "/api/cdp/connection", "cdp_host": "127.0.0.1", "cdp_port": 8700, "save_root": r"D:\_StDown\instagram", "ttl_sec": 120, "poll_sec": 1.0, "run_timeout_sec": 90, "file_logging": False, "log_keep_files": 20, "enabled": True, "page_types": {"explore": True, "single_post": True, "profile": True, "profile_reels": True}, "mime_extensions": {"image/jpeg": ".jpg", "image/jpg": ".jpg", "image/webp": ".webp", "image/avif": ".avif", "image/png": ".png"}, } try: raw = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) if isinstance(raw, dict): base.update(raw) if isinstance(raw.get("page_types"), dict): base["page_types"].update(raw["page_types"]) if isinstance(raw.get("mime_extensions"), dict): base["mime_extensions"].update(raw["mime_extensions"]) except FileNotFoundError: pass return base def save_config() -> None: tmp = CONFIG_PATH.with_suffix(".json.tmp") tmp.write_text(json.dumps(CONFIG, ensure_ascii=False, indent=2), encoding="utf-8") os.replace(tmp, CONFIG_PATH) def _cache_log_dir() -> Path: return Path(__file__).resolve().with_name("logs") def _close_file_log() -> None: global LOG_FILE, LOG_FILE_PATH with LOG_LOCK: current = LOG_FILE LOG_FILE = None LOG_FILE_PATH = "" if current is not None: try: current.flush() current.close() except Exception: pass def _prune_file_logs_before_create(root: Path, keep_files: int) -> None: # [8701 로그 파일 보관][회귀 금지] # 프로그램 시작 때 새 파일을 만들기 전에 기존 파일이 20개 이상이면 오래된 파일부터 지운다. keep = max(1, int(keep_files or 20)) while True: files = [] for path in root.glob("cache8701_log_*.log"): try: files.append((path.stat().st_mtime, path)) except Exception: continue if len(files) < keep: return files.sort(key=lambda item: item[0]) try: files[0][1].unlink() except Exception: return def _open_file_log(reason: str = "startup") -> str: global LOG_FILE, LOG_FILE_PATH with LOG_LOCK: if LOG_FILE is not None: return LOG_FILE_PATH root = _cache_log_dir() root.mkdir(parents=True, exist_ok=True) keep = max(1, int(CONFIG.get("log_keep_files") or 20)) _prune_file_logs_before_create(root, keep) stamp = datetime.now().strftime("%Y%m%d_%H%M%S") path = root / f"cache8701_log_{stamp}_pid{os.getpid()}.log" LOG_FILE = path.open("a", encoding="utf-8", buffering=1) LOG_FILE_PATH = str(path) LOG_FILE.write( f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] " f"PY_LOG_START app=cache8701 pid={os.getpid()} log={path} keep_files={keep} mode=feature_check reason={reason}\n" ) LOG_FILE.flush() return LOG_FILE_PATH def set_file_logging(enabled: bool, reason: str = "settings") -> str: CONFIG["file_logging"] = bool(enabled) if bool(enabled): return _open_file_log(reason) _close_file_log() return "" def _write_file_log(line: str) -> None: if not bool(CONFIG.get("file_logging", False)): return try: _open_file_log("lazy_start") with LOG_LOCK: if LOG_FILE is not None: LOG_FILE.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {line}\n") LOG_FILE.flush() except Exception: pass def check_log(check_id: str, status: str, text: str, key: str = "") -> None: fingerprint = f"{status}|{text}" once_key = f"{check_id}|{key}" with STATE_LOCK: if LOG_ONCE.get(once_key) == fingerprint: return LOG_ONCE[once_key] = fingerprint line = f"[캐시이미지][{check_id}][{status}] {text}" print(line, flush=True) _write_file_log(line) def log_exception(check_id: str, text: str, exc: BaseException, key: str = "") -> None: if not bool(CONFIG.get("file_logging", False)): return fingerprint = f"{type(exc).__name__}|{exc}" once_key = f"{check_id}|EXC|{key}" with STATE_LOCK: if LOG_ONCE.get(once_key) == fingerprint: return LOG_ONCE[once_key] = fingerprint detail = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)).rstrip() _write_file_log(f"[캐시이미지][{check_id}][예외추적] {text} / error={exc}\n{detail}") def normalize_url(value: Any) -> str: text = str(value or "").strip() if not text: return "" try: u = urllib.parse.urlsplit(text) return urllib.parse.urlunsplit((u.scheme.lower(), u.netloc.lower(), u.path, u.query, "")) except Exception: return text.split("#", 1)[0] def safe_name(value: Any, fallback: str = "unknown") -> str: text = re.sub(r'[\\/:*?"<>|\s]+', "_", str(value or "").strip()).strip("_.") return text[:120] or fallback def page_type_from_url(url: str) -> str: path = urllib.parse.urlsplit(str(url or "")).path.strip("/") parts = [part for part in path.split("/") if part] if not parts or parts[0] in ("explore", "search"): return "explore" if parts[0] in ("p", "reel", "reels"): return "single_post" if len(parts) >= 2 and parts[1] == "reels": return "profile_reels" return "profile" def first_value(item: dict, *keys: str) -> Any: for key in keys: value = item.get(key) if value not in (None, ""): return value return None def iter_observations(payload: dict): source_url = str(payload.get("source_url") or payload.get("url") or "") for raw in payload.get("info") or payload.get("items") or []: if not isinstance(raw, dict): continue media = raw.get("media") if isinstance(raw.get("media"), dict) else raw code = first_value(raw, "shortcode", "code") or first_value(media, "shortcode", "code") account = first_value(raw, "username", "owner_username", "account", "account_name") or first_value(media, "username", "owner_username", "account", "account_name") image_url = first_value(raw, "currentSrc", "current_src", "img_origin", "image_url", "thumbnail_url", "display_url", "img_small") or first_value(media, "currentSrc", "current_src", "img_origin", "image_url", "thumbnail_url", "display_url", "img_small") post_url = first_value(raw, "post_url", "media_url", "href") or first_value(media, "post_url", "media_url", "href") or source_url if not code or not account or not image_url: continue page_type = str(first_value(raw, "page_type") or page_type_from_url(source_url or str(post_url or ""))) metrics = { "views": first_value(media, "play_count", "view_count", "views"), "likes": first_value(media, "like_count", "likes"), "comments": first_value(media, "comment_count", "comments"), "reposts": first_value(media, "media_repost_count", "repost_count", "reposts"), "shares": first_value(media, "share_count", "shares"), "published_at": first_value(media, "created_at", "taken_at", "published_at"), "engagement_rate": first_value(media, "engagement_rate"), } yield { "account": str(account), "shortcode": str(code), "image_url": str(image_url), "post_url": str(post_url or ""), "page_type": page_type, "metrics": metrics, "tab_id": int(payload.get("tab_id") or payload.get("tabId") or 0), "run_id": str(payload.get("run_id") or payload.get("cache_run_id") or "").strip(), "observed_at": time.time(), } def cache_run_snapshot(run_id: str) -> dict: key = str(run_id or "").strip() now = time.time() timed_out_pending = 0 with STATE_LOCK: source = CACHE_RUNS.get(key) if key else None if source: target = int(source.get("target") or 0) processed = int(source.get("processed") or 0) ttl = max(10.0, float(CONFIG.get("run_timeout_sec") or 90.0)) created_at = float(source.get("created_at") or source.get("updated_at") or now) pending = max(0, target - processed) if pending and now - created_at >= ttl: timed_out_pending = pending source["failed"] = int(source.get("failed") or 0) + pending source["processed"] = target source["updated_at"] = now item = dict(source) else: item = {} if timed_out_pending: tab_id = int(item.get("tab_id") or 0) if item else 0 check_log("CI-03", "체크실패", f"CDP 이미지 URL 매칭 시간초과 / tab_id={tab_id or '-'} run_id={key} 미처리={timed_out_pending}", key or "run_timeout") item.setdefault("run_id", key) item.setdefault("target", 0) item.setdefault("processed", 0) item.setdefault("new_saved", 0) item.setdefault("duplicate_skipped", 0) item.setdefault("failed", 0) item.pop("shortcodes", None) item.pop("processed_shortcodes", None) item["pending"] = max(0, int(item["target"]) - int(item["processed"])) item["complete"] = bool(int(item["target"]) == 0 or int(item["processed"]) >= int(item["target"])) return item def cache_run_register_target(obs: dict) -> None: run_id = str(obs.get("run_id") or "").strip() if not run_id: return code = str(obs.get("shortcode") or "").strip() with STATE_LOCK: item = CACHE_RUNS.setdefault(run_id, {"run_id": run_id, "tab_id": int(obs.get("tab_id") or 0), "target": 0, "processed": 0, "new_saved": 0, "duplicate_skipped": 0, "failed": 0, "shortcodes": set(), "created_at": time.time(), "updated_at": time.time()}) codes = item.setdefault("shortcodes", set()) if code and code not in codes: codes.add(code) item["target"] = int(item.get("target") or 0) + 1 item["updated_at"] = time.time() def cache_run_record(obs: dict, result: dict) -> None: run_id = str(obs.get("run_id") or "").strip() if not run_id: return code = str(obs.get("shortcode") or "").strip() with STATE_LOCK: item = CACHE_RUNS.setdefault(run_id, {"run_id": run_id, "tab_id": int(obs.get("tab_id") or 0), "target": 0, "processed": 0, "new_saved": 0, "duplicate_skipped": 0, "failed": 0, "shortcodes": set(), "processed_shortcodes": set(), "created_at": time.time(), "updated_at": time.time()}) processed_codes = item.setdefault("processed_shortcodes", set()) if code and code in processed_codes: return if code: processed_codes.add(code) item["processed"] = int(item.get("processed") or 0) + 1 if not result.get("ok"): item["failed"] = int(item.get("failed") or 0) + 1 elif result.get("duplicate"): item["duplicate_skipped"] = int(item.get("duplicate_skipped") or 0) + 1 else: item["new_saved"] = int(item.get("new_saved") or 0) + 1 item["updated_at"] = time.time() def accept_observation(obs: dict) -> bool: if not CONFIG.get("enabled", True): return False page_type = str(obs.get("page_type") or "") if not bool((CONFIG.get("page_types") or {}).get(page_type, False)): return False key = normalize_url(obs.get("image_url")) if not key: return False tab_id = int(obs.get("tab_id") or 0) observed_key = f"{tab_id}|{key}" with STATE_LOCK: OBSERVED[observed_key] = obs cache_run_register_target(obs) check_log("CI-01", "체크완료", f"메타데이터 수신 / tab_id={tab_id or '-'} 계정={obs['account']} shortcode={obs['shortcode']} page={page_type} run_id={obs.get('run_id') or '-'}", observed_key) return True def purge_old() -> None: cutoff = time.time() - float(CONFIG.get("ttl_sec") or 120) with STATE_LOCK: for table in (OBSERVED, SAVED): for key, value in list(table.items()): ts = value if isinstance(value, (int, float)) else float(value.get("observed_at") or value.get("ts") or 0) if ts < cutoff: table.pop(key, None) for key, value in list(CACHE_RUNS.items()): if float((value or {}).get("updated_at") or 0) < time.time() - 3600: CACHE_RUNS.pop(key, None) class SimpleCDPWebSocket: def __init__(self, ws_url: str, timeout: float = 5.0): self.ws_url = ws_url self.timeout = timeout self.sock: Optional[socket.socket] = None self._id = 0 self._send_lock = threading.Lock() self._backlog: list[dict] = [] def connect(self) -> None: parsed = urllib.parse.urlparse(self.ws_url) host, port = parsed.hostname or "127.0.0.1", int(parsed.port or 80) path = parsed.path or "/" if parsed.query: path += "?" + parsed.query key = base64.b64encode(os.urandom(16)).decode("ascii") sock = socket.create_connection((host, port), timeout=self.timeout) request = ( f"GET {path} HTTP/1.1\r\nHost: {host}:{port}\r\nUpgrade: websocket\r\n" f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n" ).encode("ascii") sock.sendall(request) raw = b"" while b"\r\n\r\n" not in raw: chunk = sock.recv(4096) if not chunk: break raw += chunk if " 101 " not in raw.decode("latin1", "ignore").split("\r\n", 1)[0]: sock.close() raise RuntimeError("WebSocket handshake 실패") self.sock = sock sock.settimeout(1.0) def close(self) -> None: try: if self.sock: self.sock.close() finally: self.sock = None def _read_exact(self, size: int) -> bytes: if not self.sock: raise RuntimeError("websocket not connected") out = b"" while len(out) < size: chunk = self.sock.recv(size - len(out)) if not chunk: raise RuntimeError("websocket closed") out += chunk return out def send_text(self, text: str) -> None: if not self.sock: raise RuntimeError("websocket not connected") payload = text.encode("utf-8") header = bytearray([0x81]) length = len(payload) if length < 126: header.append(0x80 | length) elif length < 65536: header.extend([0x80 | 126]) header.extend(struct.pack("!H", length)) else: header.extend([0x80 | 127]) header.extend(struct.pack("!Q", length)) mask = os.urandom(4) masked = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload)) with self._send_lock: self.sock.sendall(bytes(header) + mask + masked) def recv_text(self, timeout: float = 1.0) -> Optional[str]: if not self.sock: raise RuntimeError("websocket not connected") self.sock.settimeout(timeout) try: first = self._read_exact(2) except socket.timeout: return None b1, b2 = first opcode = b1 & 0x0F length = b2 & 0x7F if length == 126: length = struct.unpack("!H", self._read_exact(2))[0] elif length == 127: length = struct.unpack("!Q", self._read_exact(8))[0] mask = self._read_exact(4) if b2 & 0x80 else b"" payload = self._read_exact(length) if length else b"" if mask: payload = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload)) if opcode == 0x8: raise RuntimeError("websocket closed") if opcode == 0x9: return None return payload.decode("utf-8", "replace") if opcode == 0x1 else None def recv_json(self, timeout: float = 1.0) -> dict: if self._backlog: return self._backlog.pop(0) text = self.recv_text(timeout) return json.loads(text) if text else {} def keep_message(self, data: dict) -> None: if isinstance(data, dict) and data: self._backlog.append(data) def send_cmd(self, method: str, params: Optional[dict] = None, session_id: str = "") -> int: self._id += 1 command_id = self._id payload = {"id": command_id, "method": method, "params": params or {}} if session_id: payload["sessionId"] = session_id self.send_text(json.dumps(payload, ensure_ascii=False)) return command_id def http_json(url: str, timeout: float = 2.0) -> dict: request = urllib.request.Request(url, headers={"User-Agent": "Cache8701/v304"}, method="GET") with urllib.request.urlopen(request, timeout=timeout) as response: payload = json.loads(response.read().decode("utf-8")) if not isinstance(payload, dict): raise RuntimeError("JSON object 응답 아님") return payload def auto8700_connection_url() -> str: host = str(CONFIG.get("auto8700_host") or "127.0.0.1") port = int(CONFIG.get("auto8700_port") or 8707) path = str(CONFIG.get("auto8700_connection_path") or "/api/cdp/connection") if not path.startswith("/"): path = "/" + path return f"http://{host}:{port}{path}" def fetch_auto8700_connection() -> dict: payload = http_json(auto8700_connection_url(), timeout=2.0) port = int(payload.get("cdp_port") or 0) if port <= 0: raise RuntimeError("Auto8700 CDP 포트 없음") payload["cdp_host"] = str(payload.get("cdp_host") or "127.0.0.1") payload["cdp_port"] = port payload["owner_generation"] = int(payload.get("owner_generation") or 0) payload["targets"] = [item for item in payload.get("targets") or [] if isinstance(item, dict) and item.get("id")] return payload def browser_websocket_url(connection: dict) -> str: host = str(connection.get("cdp_host") or "127.0.0.1") port = int(connection.get("cdp_port") or 0) payload = http_json(f"http://{host}:{port}/json/version", timeout=2.0) ws_url = str(payload.get("webSocketDebuggerUrl") or "").strip() if not ws_url: raise RuntimeError("browser WebSocket URL 없음") return ws_url def wait_result(cdp: SimpleCDPWebSocket, command_id: int, timeout: float = 5.0) -> dict: end = time.time() + timeout while time.time() < end: text = cdp.recv_text(timeout=min(1.0, max(0.05, end - time.time()))) if not text: continue data = json.loads(text) if int(data.get("id") or 0) == command_id: return data cdp.keep_message(data) return {"error": {"message": "command_timeout"}} def attach_target_session(cdp: SimpleCDPWebSocket, target_id: str) -> str: command_id = cdp.send_cmd("Target.attachToTarget", {"targetId": target_id, "flatten": True}) response = wait_result(cdp, command_id, timeout=5.0) session_id = str((response.get("result") or {}).get("sessionId") or "") if not session_id: raise RuntimeError(str((response.get("error") or {}).get("message") or "attach session 실패")) enable_id = cdp.send_cmd( "Network.enable", {"maxTotalBufferSize": 104857600, "maxResourceBufferSize": 10485760}, session_id=session_id, ) enabled = wait_result(cdp, enable_id, timeout=5.0) if enabled.get("error"): raise RuntimeError(str((enabled.get("error") or {}).get("message") or "Network.enable 실패")) return session_id def extension_for(mime: str, raw: bytes) -> str: if raw.startswith(b"\xff\xd8\xff"): return ".jpg" if raw.startswith(b"\x89PNG\r\n\x1a\n"): return ".png" if raw[:4] == b"RIFF" and raw[8:12] == b"WEBP": return ".webp" if b"ftypavif" in raw[:32]: return ".avif" mime = str(mime or "").split(";", 1)[0].lower() return str((CONFIG.get("mime_extensions") or {}).get(mime) or "") def save_body(obs: dict, mime: str, raw: bytes, source_url: str, cache_source: str) -> dict: extension = extension_for(mime, raw) if not extension or len(raw) < 64: return {"ok": False, "error": "unsupported_or_empty_image"} import sys module_root = Path(__file__).resolve().parents[1] if str(module_root) not in sys.path: sys.path.insert(0, str(module_root)) from Shared.storage.sort_storage import build_instagram_media_path, ensure_instagram_account_index, record_downloaded_file try: account = str(obs.get("account") or "").strip().lstrip("@") shortcode = str(obs.get("shortcode") or "").strip() target = build_instagram_media_path(ROOT, account, shortcode, extension) target.parent.mkdir(parents=True, exist_ok=True) duplicate = bool(target.is_file() and target.stat().st_size > 0) if not duplicate: part = target.with_suffix(target.suffix + ".part") part.write_bytes(raw) os.replace(part, target) index_path = ensure_instagram_account_index(ROOT, account) except Exception as exc: return {"ok": False, "error": f"file_write_failed:{exc}"} try: event_uuid = str(uuid.uuid5(uuid.NAMESPACE_URL, f"cache8701:{account.lower()}:{shortcode}:{extension}")) result = record_downloaded_file( st_root=ROOT, file_path=target, source_mode="CACHE8701_CDP", shortcode=shortcode, owner_username=account, asset_kind="IMAGE", media_index=0, source_url=source_url, post_url=str(obs.get("post_url") or ""), thumbnail_url=source_url, published_at=(obs.get("metrics") or {}).get("published_at"), file_size=int(target.stat().st_size), sha256="", md5="", account_path=account, account_folder_merge=False, batch_key=f"cache8701-{shortcode}", requested_count=1, event_uuid=event_uuid, payload={ "capture_source": "Cache8701_CDP_existing_response", "cache_source": cache_source, "metrics": obs.get("metrics") or {}, "file_contract": "{shortcode}.{actual_extension}", "duplicate_policy": "shortcode_file_exists", "hash_generation": False, }, ) return { "ok": True, "duplicate": duplicate, "file_path": str(target), "index_path": str(index_path), "bytes": int(target.stat().st_size), "registered": True, "register_error": "", "db_path": str(result.get("db_path") or ""), "post_id": int(result.get("post_id") or 0), "asset_id": int(result.get("asset_id") or 0), "file_name": str(result.get("file_name") or target.name), "relative_path": str(result.get("relative_path") or ""), "idempotent": bool(result.get("idempotent")), } except Exception as exc: return { "ok": True, "duplicate": duplicate, "file_path": str(target), "index_path": str(index_path), "bytes": int(target.stat().st_size), "registered": False, "register_error": str(exc), } def connection_can_watch(connection: dict) -> bool: # [Cache Image target 판정][회귀 금지] # OWNER active는 화면 포커스 상태다. CDP 연결 가능 여부는 cdp_alive와 Instagram target 존재로만 판단한다. return bool(connection.get("cdp_alive") and (connection.get("targets") or [])) def connection_signature(connection: dict) -> str: targets = sorted(str(item.get("id") or "") for item in connection.get("targets") or [] if item.get("id")) return json.dumps( { "host": str(connection.get("cdp_host") or "127.0.0.1"), "port": int(connection.get("cdp_port") or 0), "owner": str(connection.get("owner_target_id") or ""), "generation": int(connection.get("owner_generation") or 0), "targets": targets, }, sort_keys=True, ) def _account_from_instagram_url(url: str) -> str: try: parts = [p for p in urllib.parse.urlsplit(str(url or "")).path.split("/") if p] if not parts or parts[0] in {"explore", "search", "p", "reel", "reels", "stories"}: return "" return str(parts[0]).lower() except Exception: return "" def _observation_for_session_url(tab_id: int, url: str, session_url: str) -> Optional[dict]: normalized = normalize_url(url) with STATE_LOCK: exact = OBSERVED.get(f"{int(tab_id or 0)}|{normalized}") if exact: return exact candidates = [obs for key, obs in OBSERVED.items() if key.endswith("|" + normalized)] if not candidates: return None session_account = _account_from_instagram_url(session_url) if session_account: matched = [obs for obs in candidates if str(obs.get("account") or "").lower().lstrip("@") == session_account] if len(matched) == 1: return matched[0] return candidates[0] if len(candidates) == 1 else None def session_watcher(connection: dict, local_stop: threading.Event) -> None: cdp: Optional[SimpleCDPWebSocket] = None sessions: Dict[str, dict] = {} requests: Dict[tuple[str, str], dict] = {} loaded: Dict[str, list[dict]] = {} pending: Dict[int, tuple[dict, dict]] = {} owner_target_id = str(connection.get("owner_target_id") or "") try: ws_url = browser_websocket_url(connection) cdp = SimpleCDPWebSocket(ws_url) cdp.connect() targets = list(connection.get("targets") or []) targets.sort(key=lambda item: 0 if str(item.get("id") or "") == owner_target_id else 1) for target in targets: target_id = str(target.get("id") or "") if not target_id: continue try: session_id = attach_target_session(cdp, target_id) sessions[session_id] = {"target_id": target_id, "url": str(target.get("url") or ""), "tab_id": int(target.get("tab_id") or target.get("tabId") or 0)} check_log( "CI-02", "체크완료", f"독립 CDP 세션 연결 / tab_id={sessions[session_id].get('tab_id') or '-'} target={target_id} session={session_id[:12]} owner={target_id == owner_target_id} url={sessions[session_id].get('url') or '-'}", target_id, ) except Exception as exc: check_log("CI-02", "체크실패", f"CDP 세션 연결 실패 / target={target_id} error={exc}", target_id) if not sessions: raise RuntimeError("연결된 Instagram CDP session 없음") with STATE_LOCK: RUNTIME["sessions"] = dict(sessions) RUNTIME["last_error"] = "" LOG_ONCE.pop("CI-02|session_watcher", None) LOG_ONCE.pop("CI-02-E01|EXC|session_watcher", None) while not STOP.is_set() and not local_stop.is_set(): data = cdp.recv_json(1.0) method = str(data.get("method") or "") params = data.get("params") or {} session_id = str(data.get("sessionId") or "") if method == "Network.responseReceived" and session_id in sessions: response = params.get("response") or {} mime = str(response.get("mimeType") or "").split(";", 1)[0].lower() request_id = str(params.get("requestId") or "") if request_id and str(params.get("type") or "") == "Image" and mime in (CONFIG.get("mime_extensions") or {}): url = normalize_url(response.get("url")) requests[(session_id, request_id)] = { "url": url, "mime": mime, "source": "disk_cache" if response.get("fromDiskCache") else "prefetch_cache" if response.get("fromPrefetchCache") else "service_worker" if response.get("fromServiceWorker") else "network_or_memory", "ts": time.time(), "session_id": session_id, } elif method == "Network.loadingFinished" and session_id in sessions: request_id = str(params.get("requestId") or "") item = requests.pop((session_id, request_id), None) if item: item["request_id"] = request_id item["loaded_at"] = time.time() loaded.setdefault(item["url"], []).append(item) elif int(data.get("id") or 0) in pending: item, obs = pending.pop(int(data.get("id") or 0)) result = data.get("result") or {} if not isinstance(result.get("body"), str): failed = {"ok": False, "error": "response_body_missing"} cache_run_record(obs, failed) check_log( "CI-04", "체크실패", f"응답 body 없음 / tab_id={obs.get('tab_id') or '-'} shortcode={obs['shortcode']} error={(data.get('error') or {}).get('message', 'unknown')}", item["url"], ) continue raw = base64.b64decode(result["body"], validate=False) if result.get("base64Encoded") else result["body"].encode("latin1") check_log("CI-04", "체크완료", f"이미지 body 확보 / tab_id={obs.get('tab_id') or '-'} shortcode={obs['shortcode']} bytes={len(raw)}", item["url"]) saved = save_body(obs, item["mime"], raw, item["url"], item["source"]) cache_run_record(obs, saved) if saved.get("ok"): save_key = f"{obs.get('shortcode')}|{item['url']}" with STATE_LOCK: SAVED[save_key] = time.time() check_log("CI-05", "체크완료", f"파일 저장 / tab_id={obs.get('tab_id') or '-'} path={saved.get('file_path')} duplicate={saved.get('duplicate', False)}", item["url"]) if saved.get("registered"): check_log( "CI-06", "체크완료", f"SQLite 등록 / tab_id={obs.get('tab_id') or '-'} db={Path(str(saved.get('db_path') or '')).name} post_id={saved.get('post_id')} asset_id={saved.get('asset_id')} file={saved.get('file_name')}", item["url"], ) elif saved.get("register_error"): check_log("CI-06", "체크실패", f"SQLite 등록 실패 / {saved['register_error']}", item["url"]) else: check_log("CI-05", "체크실패", f"파일 저장 실패 / {saved.get('error')}", item["url"]) cutoff = time.time() - float(CONFIG.get("ttl_sec") or 120) for url, items in list(loaded.items()): items[:] = [item for item in items if float(item.get("loaded_at") or 0) >= cutoff] if not items: loaded.pop(url, None) continue session_meta = sessions.get(str(items[0].get("session_id") or ""), {}) if items else {} observation = _observation_for_session_url(int(session_meta.get("tab_id") or 0), url, str(session_meta.get("url") or "")) if not observation: continue item = items.pop(0) if not items: loaded.pop(url, None) check_log("CI-03", "체크완료", f"CDP URL 매칭 / tab_id={observation.get('tab_id') or '-'} shortcode={observation['shortcode']} source={item['source']} run_id={observation.get('run_id') or '-'}", url) command_id = cdp.send_cmd( "Network.getResponseBody", {"requestId": item["request_id"]}, session_id=item["session_id"], ) pending[command_id] = (item, observation) except Exception as exc: with STATE_LOCK: RUNTIME["last_error"] = str(exc) check_log("CI-02", "체크실패", f"독립 CDP 세션 종료 / error={exc}", "session_watcher") log_exception("CI-02-E01", "독립 CDP 세션 예외", exc, "session_watcher") finally: if cdp: cdp.close() with STATE_LOCK: RUNTIME["sessions"] = {} def stop_runtime_watcher() -> None: with STATE_LOCK: event = RUNTIME.get("watcher_stop") thread = RUNTIME.get("watcher") RUNTIME["watcher_stop"] = None RUNTIME["watcher"] = None RUNTIME["watcher_signature"] = "" if isinstance(event, threading.Event): event.set() if isinstance(thread, threading.Thread) and thread.is_alive(): thread.join(timeout=2.0) def target_manager() -> None: check_log("CI-00", "체크시작", f"Cache8701 시작 / Auto8700={auto8700_connection_url()}") while not STOP.is_set(): purge_old() try: connection = fetch_auto8700_connection() with STATE_LOCK: RUNTIME["connection"] = dict(connection) port = int(connection.get("cdp_port") or 0) owner = str(connection.get("owner_target_id") or "") generation = int(connection.get("owner_generation") or 0) if not connection.get("cdp_alive"): raise RuntimeError(f"Chrome CDP 준비 안 됨 / port={port}") # [Cache Image 기존 연결 보호][회귀 금지] # OWNER active=false는 사용자가 다른 탭을 보고 있다는 뜻일 뿐 CDP target 단절이 아니다. # Instagram target이 존재하면 비활성 OWNER를 포함한 전체 target 감시를 계속한다. if not connection_can_watch(connection): raise RuntimeError(f"Instagram target 없음 / port={port}") signature = connection_signature(connection) with STATE_LOCK: LOG_ONCE.pop("CI-00|connection_wait", None) LOG_ONCE.pop("CI-00-E01|EXC|connection_wait", None) check_log( "CI-00", "체크완료", f"연결정보 확인 / cdp_port={port} owner={owner} owner_active={bool(connection.get('owner_active'))} generation={generation} targets={len(connection.get('targets') or [])}", signature, ) with STATE_LOCK: current_signature = str(RUNTIME.get("watcher_signature") or "") current_thread = RUNTIME.get("watcher") if signature != current_signature or not isinstance(current_thread, threading.Thread) or not current_thread.is_alive(): stop_runtime_watcher() local_stop = threading.Event() thread = threading.Thread( target=session_watcher, args=(connection, local_stop), daemon=True, name=f"Cache8701-CDP-{generation}", ) with STATE_LOCK: RUNTIME["watcher_signature"] = signature RUNTIME["watcher_stop"] = local_stop RUNTIME["watcher"] = thread thread.start() except Exception as exc: stop_runtime_watcher() with STATE_LOCK: RUNTIME["last_error"] = str(exc) check_log("CI-00", "체크실패", f"Auto8700 CDP 연결정보 대기 / {exc}", "connection_wait") log_exception("CI-00-E01", "Auto8700 연결정보 또는 target manager 예외", exc, "connection_wait") STOP.wait(float(CONFIG.get("poll_sec") or 1.0)) stop_runtime_watcher() class Handler(BaseHTTPRequestHandler): def log_message(self, fmt, *args): return def cors(self) -> None: self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.send_header("Access-Control-Allow-Methods", "GET,POST,OPTIONS") def send_json(self, status: int, payload: dict) -> None: raw = json.dumps(payload, ensure_ascii=False).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 do_OPTIONS(self) -> None: self.send_response(204) self.cors() self.end_headers() def do_GET(self) -> None: if self.path.startswith("/api/status"): query = urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query) run_id = str((query.get("run_id") or [""])[0]).strip() return self.send_json(200, {"ok": True, "cache": cache_run_snapshot(run_id)}) if self.path.startswith("/api/health"): with STATE_LOCK: connection = dict(RUNTIME.get("connection") or {}) payload = { "ok": True, "version": CONFIG.get("version"), "port": CONFIG.get("port"), "observed": len(OBSERVED), "sessions": len(RUNTIME.get("sessions") or {}), "saved_recent": len(SAVED), "cdp_port": connection.get("cdp_port"), "owner_target_id": connection.get("owner_target_id"), "owner_generation": connection.get("owner_generation"), "last_error": RUNTIME.get("last_error") or "", "file_logging": bool(CONFIG.get("file_logging", False)), "log_file": LOG_FILE_PATH, } return self.send_json(200, payload) return self.send_json(404, {"ok": False, "error": "not_found"}) def do_POST(self) -> None: try: length = int(self.headers.get("Content-Length") or 0) payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}") except Exception as exc: return self.send_json(400, {"ok": False, "error": str(exc)}) if self.path.startswith("/api/meta"): observations = list(iter_observations(payload)) accepted = sum(1 for observation in observations if accept_observation(observation)) run_id = str(payload.get("run_id") or payload.get("cache_run_id") or "").strip() return self.send_json(200, {"ok": True, "accepted": accepted, "cache": cache_run_snapshot(run_id)}) if self.path.startswith("/api/settings"): settings = payload.get("settings") if isinstance(payload.get("settings"), dict) else payload CONFIG["enabled"] = bool(settings.get("auto_thumbnail_enabled", CONFIG.get("enabled", True))) logging_value = bool(settings.get("cache8701_file_logging", CONFIG.get("file_logging", False))) CONFIG["page_types"] = { "explore": bool(settings.get("auto_thumbnail_explore", False)), "single_post": bool(settings.get("auto_thumbnail_single_post", True)), "profile": bool(settings.get("auto_thumbnail_profile", True)), "profile_reels": bool(settings.get("auto_thumbnail_profile_reels", True)), } set_file_logging(logging_value, "settings_change") save_config() check_log("CI-00", "체크완료", f"설정 반영 / enabled={CONFIG['enabled']} logging={CONFIG['file_logging']} log_file={LOG_FILE_PATH or '-'} pages={CONFIG['page_types']}", "settings") return self.send_json(200, {"ok": True, "settings": {"enabled": CONFIG["enabled"], "file_logging": CONFIG["file_logging"], "log_file": LOG_FILE_PATH, "page_types": CONFIG["page_types"]}}) return self.send_json(404, {"ok": False, "error": "not_found"}) def main() -> None: global CONFIG parser = argparse.ArgumentParser() parser.add_argument("--host") parser.add_argument("--port", type=int) parser.add_argument("--cdp-port", type=int, help="진단용 강제값. 정상 실행은 Auto8700 연결정보를 사용한다.") args = parser.parse_args() CONFIG = load_config() if args.host: CONFIG["host"] = args.host if args.port: CONFIG["port"] = args.port if args.cdp_port: CONFIG["cdp_port"] = args.cdp_port if bool(CONFIG.get("file_logging", False)): set_file_logging(True, "program_start") threading.Thread(target=target_manager, daemon=True, name="Cache8701TargetManager").start() server = ThreadingHTTPServer((str(CONFIG.get("host") or "127.0.0.1"), int(CONFIG.get("port") or 8701)), Handler) check_log("CI-00", "체크완료", f"HTTP 준비 / http://{CONFIG['host']}:{CONFIG['port']}") try: server.serve_forever() except KeyboardInterrupt: pass finally: STOP.set() server.server_close() _close_file_log() if __name__ == "__main__": main()