import base64 import cgi import hashlib import hmac import html import json import mimetypes import os import re import secrets import shutil import sqlite3 import time import urllib.parse import urllib.request import zipfile from http import HTTPStatus from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from email.utils import formatdate from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DEPLOY_ROOT = ROOT.parent if ROOT.name.lower() == "app" else ROOT DATA_DIR = Path(os.environ.get("DPSTAR_DATA_DIR", str(DEPLOY_ROOT / "data"))).expanduser().resolve() UPLOAD_DIR = Path(os.environ.get("DPSTAR_UPLOAD_DIR", str(DEPLOY_ROOT / "uploads"))).expanduser().resolve() DB_PATH = DATA_DIR / "dpstar.sqlite3" MENU_CONFIG_PATH = DATA_DIR / "bottom_menu.json" PROFILE_MENU_CONFIG_PATH = DATA_DIR / "profile_menu.json" ACTION_BUTTON_CONFIG_PATH = DATA_DIR / "action_buttons.json" PRIVACY_POLICY_PATH = DATA_DIR / "privacy_policy.json" USER_AGREEMENT_PATH = DATA_DIR / "user_agreement.json" PRIVACY_DOCX_IMPORT_PATH = Path.home() / "Desktop" / "AI衣橱搭配师隐私政策.docx" HOST = os.environ.get("DPSTAR_HOST", "127.0.0.1") PORT = int(os.environ.get("DPSTAR_PORT", "8000")) MAX_UPLOAD_BYTES = 8 * 1024 * 1024 ORPHAN_UPLOAD_TTL_SECONDS = 3600 ALLOWED_IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".webp"} SENSITIVE_TERMS = [ "色情", "裸聊", "约炮", "嫖娼", "卖淫", "博彩", "赌博", "毒品", "枪支", "炸药", "管制刀具", "杀人", "自杀", "恐怖", "暴恐", "诈骗", "洗钱", "非法集资", "假证", "黑客", "木马", "反动", "邪教", ] def ensure_dirs(): DATA_DIR.mkdir(exist_ok=True) UPLOAD_DIR.mkdir(exist_ok=True) def default_bottom_menu(): return [ {"id": "today", "label": "今日推荐", "target": "today", "icon": "today", "iconUrl": "", "visible": True, "sort": 1}, {"id": "closet", "label": "衣橱", "target": "closet", "icon": "closet", "iconUrl": "", "visible": True, "sort": 2}, {"id": "profile", "label": "我的", "target": "profile", "icon": "profile", "iconUrl": "", "visible": True, "sort": 3}, ] def default_profile_menu(): return [ {"id": "profile_info", "label": "我的资料", "target": "profile", "url": "", "icon": "plus", "iconUrl": "", "visible": True, "sort": 1}, {"id": "weekly_saved", "label": "保存的本周穿搭", "target": "weekly", "url": "", "icon": "calendar", "iconUrl": "", "visible": True, "sort": 2}, {"id": "generation_stats", "label": "我的穿搭生成数据", "target": "generationStats", "url": "", "icon": "star", "iconUrl": "", "visible": True, "sort": 3}, {"id": "service", "label": "关注服务号", "target": "url", "url": "", "icon": "wechat", "iconUrl": "", "visible": True, "sort": 7}, {"id": "privacy", "label": "隐私条款", "target": "privacy", "url": "", "icon": "shield", "iconUrl": "", "visible": True, "sort": 8}, {"id": "agreement", "label": "用户协议", "target": "agreement", "url": "", "icon": "card", "iconUrl": "", "visible": True, "sort": 9}, {"id": "logout", "label": "退出登录", "target": "logout", "url": "", "icon": "logout", "iconUrl": "", "visible": True, "sort": 9}, {"id": "delete_account", "label": "注销账户", "target": "deleteAccount", "url": "", "icon": "shield", "iconUrl": "", "visible": True, "sort": 10}, ] def default_action_buttons(): return [ {"id": "generate_today", "label": "生成今日穿搭", "iconUrl": "", "visible": True, "sort": 1}, {"id": "generate_week", "label": "生成一周穿搭", "iconUrl": "", "visible": True, "sort": 2}, {"id": "new_cloth", "label": "拍照录入新的衣服", "iconUrl": "", "visible": True, "sort": 3}, {"id": "analyze_closet", "label": "生成衣橱分析", "iconUrl": "", "visible": True, "sort": 4}, {"id": "upload_photo", "label": "上传/替换照片", "iconUrl": "", "visible": True, "sort": 5}, {"id": "batch_upload_photo", "label": "批量上传照片（上限10张）", "iconUrl": "", "visible": True, "sort": 6}, {"id": "save_cloth", "label": "录入到我的衣橱里", "iconUrl": "", "visible": True, "sort": 7}, {"id": "save_today", "label": "保存今日穿搭", "iconUrl": "", "visible": True, "sort": 8}, {"id": "continue_today", "label": "继续推荐新的穿搭", "iconUrl": "", "visible": True, "sort": 9}, {"id": "save_week", "label": "保存本周穿搭", "iconUrl": "", "visible": True, "sort": 10}, {"id": "reroll_week", "label": "重新推荐本周穿搭", "iconUrl": "", "visible": True, "sort": 11}, {"id": "delete_saved_week", "label": "删除本周穿搭", "iconUrl": "", "visible": True, "sort": 12}, {"id": "clear_recent_outfits", "label": "删除近7套生成穿搭的记录", "iconUrl": "", "visible": True, "sort": 13}, {"id": "back_to_top", "label": "回顶部", "iconUrl": "", "visible": True, "sort": 14}, ] def read_bottom_menu(): ensure_dirs() if not MENU_CONFIG_PATH.exists(): write_bottom_menu(default_bottom_menu()) try: data = json.loads(MENU_CONFIG_PATH.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): data = default_bottom_menu() return normalize_bottom_menu(data) def read_profile_menu(): ensure_dirs() if not PROFILE_MENU_CONFIG_PATH.exists(): write_profile_menu(default_profile_menu()) try: data = json.loads(PROFILE_MENU_CONFIG_PATH.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): data = default_profile_menu() if isinstance(data, dict): data = data.get("items") or data.get("profileMenu") or data.get("value") or [] return normalize_profile_menu(data) def read_action_buttons(): ensure_dirs() if not ACTION_BUTTON_CONFIG_PATH.exists(): write_action_buttons(default_action_buttons()) try: data = json.loads(ACTION_BUTTON_CONFIG_PATH.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): data = default_action_buttons() return normalize_action_buttons(data) def write_bottom_menu(items): ensure_dirs() MENU_CONFIG_PATH.write_text(json.dumps(normalize_bottom_menu(items), ensure_ascii=False, indent=2), encoding="utf-8") def write_profile_menu(items): ensure_dirs() PROFILE_MENU_CONFIG_PATH.write_text(json.dumps(normalize_profile_menu(items), ensure_ascii=False, indent=2), encoding="utf-8") def write_action_buttons(items): ensure_dirs() ACTION_BUTTON_CONFIG_PATH.write_text(json.dumps(normalize_action_buttons(items), ensure_ascii=False, indent=2), encoding="utf-8") def extract_docx_text(path): try: with zipfile.ZipFile(path) as docx: xml = docx.read("word/document.xml").decode("utf-8", errors="ignore") except (OSError, KeyError, zipfile.BadZipFile): return "" xml = re.sub(r"", "\n", xml) text = re.sub(r"<[^>]+>", "", xml) lines = [html.unescape(line).strip() for line in text.splitlines()] return "\n".join(line for line in lines if line) def default_privacy_policy(): imported = extract_docx_text(PRIVACY_DOCX_IMPORT_PATH) if PRIVACY_DOCX_IMPORT_PATH.exists() else "" return { "title": "AI衣橱搭配师隐私政策", "content": imported or "暂无隐私条款，请在后台上传或填写隐私政策内容。", "updatedAt": int(time.time()), } def default_user_agreement(): return { "title": "DPSTAR AI 用户协议", "content": "暂无用户协议，请在后台上传或填写用户协议内容。", "updatedAt": int(time.time()), } def read_privacy_policy(): ensure_dirs() if not PRIVACY_POLICY_PATH.exists(): write_privacy_policy(default_privacy_policy()) try: data = json.loads(PRIVACY_POLICY_PATH.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): data = default_privacy_policy() return { "title": str(data.get("title") or "AI衣橱搭配师隐私政策").strip(), "content": str(data.get("content") or "").strip(), "updatedAt": int(data.get("updatedAt") or int(time.time())), } def read_user_agreement(): ensure_dirs() if not USER_AGREEMENT_PATH.exists(): write_user_agreement(default_user_agreement()) try: data = json.loads(USER_AGREEMENT_PATH.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): data = default_user_agreement() return { "title": str(data.get("title") or "DPSTAR AI 用户协议").strip(), "content": str(data.get("content") or "").strip(), "updatedAt": int(data.get("updatedAt") or int(time.time())), } def write_privacy_policy(policy): ensure_dirs() title = str(policy.get("title") or "AI衣橱搭配师隐私政策").strip()[:80] content = str(policy.get("content") or "").strip() PRIVACY_POLICY_PATH.write_text( json.dumps({"title": title, "content": content, "updatedAt": int(time.time())}, ensure_ascii=False, indent=2), encoding="utf-8", ) def write_user_agreement(policy): ensure_dirs() title = str(policy.get("title") or "DPSTAR AI 用户协议").strip()[:80] content = str(policy.get("content") or "").strip() USER_AGREEMENT_PATH.write_text( json.dumps({"title": title, "content": content, "updatedAt": int(time.time())}, ensure_ascii=False, indent=2), encoding="utf-8", ) def normalize_bottom_menu(items): allowed_targets = {"today", "closet", "profile"} allowed_icons = {"today", "closet", "profile", "home", "star", "bag", "user"} normalized = [] for index, item in enumerate(items or []): target = item.get("target") if isinstance(item, dict) else "" if target not in allowed_targets: target = "today" label = str(item.get("label", "") if isinstance(item, dict) else "").strip()[:8] or "菜单" icon = str(item.get("icon", "") if isinstance(item, dict) else "").strip() if icon not in allowed_icons: icon = target normalized.append( { "id": str(item.get("id") or f"menu_{index + 1}")[:40], "label": label, "target": target, "icon": icon, "iconUrl": str(item.get("iconUrl", "") if isinstance(item, dict) else "").strip(), "visible": bool(item.get("visible", True) if isinstance(item, dict) else True), "sort": int(item.get("sort") or index + 1 if isinstance(item, dict) else index + 1), } ) visible_targets = {item["target"] for item in normalized if item["visible"]} if not normalized or not visible_targets: normalized = default_bottom_menu() return sorted(normalized, key=lambda item: item["sort"]) def normalize_profile_menu(items): allowed_targets = {"today", "closet", "profile", "login", "url", "privacy", "agreement", "generationStats", "weekly", "logout", "deleteAccount"} allowed_icons = {"plus", "yen", "card", "pin", "bell", "wechat", "shield", "logout", "star", "user", "bag", "calendar"} normalized = [] for index, item in enumerate(items or []): if not isinstance(item, dict): continue target = str(item.get("target") or "url").strip() if str(item.get("id") or "").strip() == "weekly_saved": target = "weekly" if target not in allowed_targets: target = "url" icon = str(item.get("icon") or "plus").strip() if icon not in allowed_icons: icon = "plus" normalized.append( { "id": str(item.get("id") or f"profile_menu_{index + 1}")[:50], "label": str(item.get("label") or "菜单").strip()[:30], "target": target, "url": str(item.get("url") or "").strip(), "icon": icon, "iconUrl": str(item.get("iconUrl") or "").strip(), "visible": bool(item.get("visible", True)), "sort": int(item.get("sort") or index + 1), } ) deduped = [] seen_keys = set() for menu_item in sorted(normalized, key=lambda item: item["sort"]): key = "weekly" if menu_item["id"] == "weekly_saved" or menu_item["target"] == "weekly" else f'{menu_item["target"]}:{menu_item["label"]}' if key in seen_keys: continue seen_keys.add(key) deduped.append(menu_item) normalized = deduped if not normalized: normalized = default_profile_menu() return sorted(normalized, key=lambda item: item["sort"]) def normalize_action_buttons(items): default_by_id = {item["id"]: item for item in default_action_buttons()} normalized = [] seen = set() for index, item in enumerate(items or []): if not isinstance(item, dict): continue item_id = str(item.get("id") or "").strip() if item_id not in default_by_id or item_id in seen: continue seen.add(item_id) default_item = default_by_id[item_id] normalized.append({ "id": item_id, "label": str(item.get("label") or default_item["label"]).strip()[:20], "iconUrl": str(item.get("iconUrl") or "").strip(), "visible": bool(item.get("visible", True)), "sort": int(item.get("sort") or default_item["sort"]), }) for item_id, default_item in default_by_id.items(): if item_id not in seen: normalized.append(default_item) return sorted(normalized, key=lambda item: item["sort"]) def db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): ensure_dirs() read_bottom_menu() read_profile_menu() with db() as conn: conn.executescript( """ create table if not exists users ( id integer primary key autoincrement, phone text, nickname text not null, avatar_url text, gender text default '男生', height text, weight text, created_at integer not null ); create table if not exists sessions ( token text primary key, user_id integer not null, created_at integer not null, foreign key(user_id) references users(id) ); create table if not exists wardrobe_items ( id integer primary key autoincrement, user_id integer not null, name text not null, category text not null, color text, style text, fit text, season text, scene text, difficulty text, image_url text, usage integer default 0, created_at integer not null, updated_at integer not null, foreign key(user_id) references users(id) ); create table if not exists admins ( id integer primary key autoincrement, username text not null unique, password_hash text not null, created_at integer not null, updated_at integer not null ); create table if not exists admin_sessions ( token text primary key, admin_id integer not null, created_at integer not null, foreign key(admin_id) references admins(id) ); create table if not exists sms_settings ( id integer primary key check (id = 1), enabled integer default 1, yunpian_api_key text, sms_signature text, sms_template text, updated_at integer ); create table if not exists sms_codes ( phone text primary key, code text not null, expires_at integer not null, created_at integer not null ); create table if not exists weekly_outfits ( user_id integer primary key, plan_json text not null, created_at integer not null, updated_at integer not null, foreign key(user_id) references users(id) ); create table if not exists generation_stats ( user_id integer not null, stat_date text not null, kind text not null, count integer default 0, updated_at integer not null, primary key(user_id, stat_date, kind), foreign key(user_id) references users(id) ); create table if not exists user_activity_stats ( user_id integer not null, stat_date text not null, active_seconds integer default 0, last_seen integer not null, updated_at integer not null, primary key(user_id, stat_date), foreign key(user_id) references users(id) ); """ ) columns = [row["name"] for row in conn.execute("pragma table_info(users)").fetchall()] def add_user_column(name, definition): if name not in columns: conn.execute(f"alter table users add column {name} {definition}") add_user_column("phone", "text") add_user_column("is_blocked", "integer default 0") add_user_column("age_info", "text") add_user_column("size", "text") add_user_column("wardrobe_limit", "integer default 50") add_user_column("preferred_style", "text") add_user_column("common_scenes", "text") add_user_column("profile_city", "text") add_user_column("problem", "text") add_user_column("today_city", "text") add_user_column("today_scene", "text") add_user_column("today_mood", "text") add_user_column("accessory_mode", "text") add_user_column("updated_at", "integer") add_user_column("password_hash", "text") admin_columns = [row["name"] for row in conn.execute("pragma table_info(admins)").fetchall()] if "permissions" not in admin_columns: conn.execute("alter table admins add column permissions text default 'all'") sms_columns = [row["name"] for row in conn.execute("pragma table_info(sms_settings)").fetchall()] if "enabled" not in sms_columns: conn.execute("alter table sms_settings add column enabled integer default 1") conn.execute("update users set wardrobe_limit = 50 where wardrobe_limit is null or wardrobe_limit <= 0") conn.execute("create unique index if not exists idx_users_phone on users(phone)") conn.execute( """ insert or ignore into sms_settings (id, enabled, yunpian_api_key, sms_signature, sms_template, updated_at) values (1, 1, '', 'DPSTAR AI', '【#signature#】您的验证码是#code#，5分钟内有效。', ?) """, (int(time.time()),), ) admin = conn.execute("select id from admins where username = ?", ("admin",)).fetchone() if not admin: now = int(time.time()) conn.execute( "insert into admins (username, password_hash, created_at, updated_at) values (?, ?, ?, ?)", ("admin", hash_password("admin123"), now, now), ) def json_response(handler, payload, status=HTTPStatus.OK): body = json.dumps(payload, ensure_ascii=False).encode("utf-8") handler.send_response(status) handler.send_header("Content-Type", "application/json; charset=utf-8") handler.send_header("Access-Control-Allow-Origin", "*") handler.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization") handler.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") handler.send_header("Content-Length", str(len(body))) handler.end_headers() handler.wfile.write(body) def read_json(handler): length = int(handler.headers.get("Content-Length", "0")) if not length: return {} return json.loads(handler.rfile.read(length).decode("utf-8")) def flatten_text(value): if isinstance(value, dict): return " ".join(flatten_text(item) for item in value.values()) if isinstance(value, list): return " ".join(flatten_text(item) for item in value) if value is None: return "" return str(value) def find_sensitive_term(value): text = flatten_text(value).lower() for term in SENSITIVE_TERMS: if term.lower() in text: return term return "" def reject_sensitive_payload(handler, payload): if configured_value("DPSTAR_SAFETY_ENABLED", "1") == "0": return False term = find_sensitive_term(payload) if term: json_response(handler, {"error": f"内容包含敏感或违法风险词：{term}，请修改后再提交"}, HTTPStatus.BAD_REQUEST) return True return False def check_image_safety(image_path): if configured_value("DPSTAR_SAFETY_ENABLED", "1") == "0": return {"allowed": True, "reason": "安全审查已关闭"} api_key = configured_value("DPSTAR_IMAGE_MODERATION_API_KEY") api_url = configured_value("DPSTAR_IMAGE_MODERATION_API_URL") model = configured_value("DPSTAR_IMAGE_MODERATION_MODEL", "qwen-vl-plus") if not api_key or not api_url: return {"allowed": True, "reason": "未配置图片安全审查接口，已跳过图片内容审查"} image_bytes = Path(image_path).read_bytes() data_url = "data:image/jpeg;base64," + base64.b64encode(image_bytes).decode("ascii") payload = { "model": model, "messages": [ { "role": "user", "content": [ {"type": "text", "text": "请审核图片是否包含色情低俗、暴恐血腥、违法违禁、政治敏感、诈骗广告等风险。只返回JSON：allowed(boolean), reason(string)。"}, {"type": "image_url", "image_url": {"url": data_url}}, ], } ], "temperature": 0, } request = urllib.request.Request( api_url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}, method="POST", ) with urllib.request.urlopen(request, timeout=60) as response: data = json.loads(response.read().decode("utf-8")) text = data.get("output_text", "") if not text: for choice in data.get("choices", []): content = choice.get("message", {}).get("content", "") text += content if isinstance(content, str) else "" if text.startswith("```"): text = text.strip("`").replace("json\n", "", 1).replace("JSON\n", "", 1).strip() if not text.startswith("{"): start = text.find("{") end = text.rfind("}") if start >= 0 and end > start: text = text[start : end + 1] result = json.loads(text) return {"allowed": bool(result.get("allowed")), "reason": result.get("reason") or "图片审核未通过"} def storage_mode(): return "oss" if configured_value("DPSTAR_STORAGE_MODE", "local").lower() == "oss" else "local" def oss_settings(): endpoint = configured_value("DPSTAR_OSS_ENDPOINT", "").strip().rstrip("/") endpoint = re.sub(r"^https?://", "", endpoint, flags=re.I) return { "endpoint": endpoint, "bucket": configured_value("DPSTAR_OSS_BUCKET", "").strip(), "access_key_id": configured_value("DPSTAR_OSS_ACCESS_KEY_ID", "").strip(), "access_key_secret": configured_value("DPSTAR_OSS_ACCESS_KEY_SECRET", "").strip(), "prefix": configured_value("DPSTAR_OSS_PREFIX", "uploads").strip().strip("/") or "uploads", "public_base_url": configured_value("DPSTAR_OSS_PUBLIC_BASE_URL", "").strip().rstrip("/"), } def oss_object_url(settings, object_key): host = settings["endpoint"] if not host.startswith(f'{settings["bucket"]}.'): host = f'{settings["bucket"]}.{host}' base_url = settings["public_base_url"] or f"https://{host}" return f'{base_url}/{urllib.parse.quote(object_key, safe="/")}' def oss_request(method, object_key, data=None, content_type=""): settings = oss_settings() required = ("endpoint", "bucket", "access_key_id", "access_key_secret") if any(not settings[key] for key in required): raise ValueError("OSS 配置不完整，请填写 Endpoint、Bucket、AccessKey ID 和 Secret") request_date = formatdate(timeval=None, localtime=False, usegmt=True) canonical_resource = f'/{settings["bucket"]}/{object_key}' string_to_sign = f"{method}\n\n{content_type}\n{request_date}\n{canonical_resource}" signature = base64.b64encode( hmac.new(settings["access_key_secret"].encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha1).digest() ).decode("ascii") headers = { "Date": request_date, "Authorization": f'OSS {settings["access_key_id"]}:{signature}', } if content_type: headers["Content-Type"] = content_type request = urllib.request.Request( oss_object_url(settings, object_key), data=data, headers=headers, method=method, ) with urllib.request.urlopen(request, timeout=60) as response: response.read() return oss_object_url(settings, object_key) def upload_file_to_oss(file_path, filename): settings = oss_settings() object_key = f'{settings["prefix"]}/{filename}' content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream" return oss_request("PUT", object_key, file_path.read_bytes(), content_type) def oss_object_key_from_url(image_url): if storage_mode() != "oss" or not image_url: return "" settings = oss_settings() parsed = urllib.parse.urlparse(str(image_url)) allowed_hosts = {urllib.parse.urlparse(settings["public_base_url"]).netloc} if settings["public_base_url"] else set() endpoint_host = settings["endpoint"] if endpoint_host: allowed_hosts.add(endpoint_host if endpoint_host.startswith(f'{settings["bucket"]}.') else f'{settings["bucket"]}.{endpoint_host}') if parsed.netloc not in allowed_hosts: return "" object_key = urllib.parse.unquote(parsed.path.lstrip("/")) return object_key if object_key.startswith(f'{settings["prefix"]}/') else "" def save_checked_upload(field, prefix): original_name = field.filename or "" if find_sensitive_term(original_name): raise ValueError("文件名包含敏感或违法风险词，请修改后再上传") suffix = Path(original_name).suffix.lower() or ".jpg" if suffix not in ALLOWED_IMAGE_SUFFIXES: raise ValueError("仅支持 jpg、jpeg、png、webp 图片格式") data = field.file.read(MAX_UPLOAD_BYTES + 1) if len(data) > MAX_UPLOAD_BYTES: raise ValueError("图片不能超过 8MB") if not data: raise ValueError("图片文件为空") filename = f"{prefix}_{int(time.time())}_{secrets.token_hex(6)}{suffix}" target = UPLOAD_DIR / filename target.write_bytes(data) try: safety = check_image_safety(target) except Exception as exc: target.unlink(missing_ok=True) raise ValueError(f"图片安全审核失败：{exc}") from exc if not safety.get("allowed"): target.unlink(missing_ok=True) raise ValueError(safety.get("reason") or "图片包含敏感或违法风险，已拒绝上传") if storage_mode() == "oss": try: url = upload_file_to_oss(target, filename) except Exception as exc: target.unlink(missing_ok=True) raise ValueError(f"上传阿里云 OSS 失败：{exc}") from exc target.unlink(missing_ok=True) return url return f"/uploads/{filename}" def upload_file_path(image_url): if not image_url: return None path = urllib.parse.urlparse(str(image_url)).path if not path.startswith("/uploads/"): return None target = (UPLOAD_DIR / Path(path).name).resolve() upload_root = UPLOAD_DIR.resolve() if upload_root not in target.parents or not target.is_file(): return None return target def image_url_is_referenced(conn, image_url): value = str(image_url or "").strip() if not value: return False if conn.execute("select 1 from wardrobe_items where image_url = ? limit 1", (value,)).fetchone(): return True if conn.execute("select 1 from users where avatar_url = ? limit 1", (value,)).fetchone(): return True if conn.execute("select 1 from weekly_outfits where plan_json like ? limit 1", (f"%{value}%",)).fetchone(): return True configured_values = [read_bottom_menu(), read_profile_menu(), read_action_buttons()] return any(value in json.dumps(item, ensure_ascii=False) for item in configured_values) def delete_oss_upload(image_url): object_key = oss_object_key_from_url(image_url) if not object_key: return oss_request("DELETE", object_key) def cleanup_upload_if_unused(conn, image_url): if image_url_is_referenced(conn, image_url): return target = upload_file_path(image_url) if target: target.unlink(missing_ok=True) return try: delete_oss_upload(image_url) except Exception as exc: print(f"OSS object cleanup failed: {exc}") def upload_url_path(value): if not value: return "" path = urllib.parse.urlparse(str(value)).path.replace("\\", "/") return path if path.startswith("/uploads/") else "" def collect_upload_paths_from_value(value, paths): if isinstance(value, dict): for item in value.values(): collect_upload_paths_from_value(item, paths) return if isinstance(value, list): for item in value: collect_upload_paths_from_value(item, paths) return if not isinstance(value, str): return direct_path = upload_url_path(value) if direct_path: paths.add(direct_path) for match in re.findall(r"/uploads/[^\s\"'<>),]+", value): path = upload_url_path(match) if path: paths.add(path) def referenced_upload_paths(conn): paths = set() for row in conn.execute("select image_url from wardrobe_items where image_url like '%/uploads/%'").fetchall(): collect_upload_paths_from_value(row["image_url"], paths) for row in conn.execute("select plan_json from weekly_outfits where plan_json like '%/uploads/%'").fetchall(): collect_upload_paths_from_value(row["plan_json"], paths) try: user_rows = conn.execute("select avatar_url from users where avatar_url like '%/uploads/%'").fetchall() except sqlite3.OperationalError: user_rows = [] for row in user_rows: collect_upload_paths_from_value(row["avatar_url"], paths) collect_upload_paths_from_value(read_bottom_menu(), paths) collect_upload_paths_from_value(read_profile_menu(), paths) collect_upload_paths_from_value(read_action_buttons(), paths) collect_upload_paths_from_value(configured_value("DPSTAR_DEFAULT_AVATAR_URL", ""), paths) collect_upload_paths_from_value(configured_value("DPSTAR_DEFAULT_MALE_AVATAR_URL", ""), paths) collect_upload_paths_from_value(configured_value("DPSTAR_DEFAULT_FEMALE_AVATAR_URL", ""), paths) return paths def cleanup_orphan_uploads(conn, grace_seconds=ORPHAN_UPLOAD_TTL_SECONDS): ensure_dirs() referenced = referenced_upload_paths(conn) now = time.time() deleted = 0 for target in UPLOAD_DIR.glob("*"): if not target.is_file(): continue upload_path = f"/uploads/{target.name}" if upload_path in referenced: continue try: if now - target.stat().st_mtime < grace_seconds: continue target.unlink() deleted += 1 except OSError: continue existing = len([path for path in UPLOAD_DIR.glob("*") if path.is_file()]) return {"existingImages": existing, "deletedImages": deleted} def upload_gallery_payload(): ensure_dirs() images = [] for target in sorted(UPLOAD_DIR.glob("*"), key=lambda path: path.stat().st_mtime if path.is_file() else 0, reverse=True): if not target.is_file(): continue try: stat = target.stat() except OSError: continue images.append({ "name": target.name, "url": f"/uploads/{target.name}", "size": stat.st_size, "updatedAt": int(stat.st_mtime), }) return images def delete_users_and_uploads(conn, user_ids): if not user_ids: return placeholders = ",".join("?" for _ in user_ids) phones = [ row["phone"] for row in conn.execute(f"select phone from users where id in ({placeholders})", user_ids).fetchall() if row["phone"] ] image_rows = conn.execute(f"select image_url from wardrobe_items where user_id in ({placeholders})", user_ids).fetchall() conn.execute(f"delete from sessions where user_id in ({placeholders})", user_ids) conn.execute(f"delete from weekly_outfits where user_id in ({placeholders})", user_ids) conn.execute(f"delete from generation_stats where user_id in ({placeholders})", user_ids) conn.execute(f"delete from wardrobe_items where user_id in ({placeholders})", user_ids) if phones: phone_placeholders = ",".join("?" for _ in phones) conn.execute(f"delete from sms_codes where phone in ({phone_placeholders})", phones) conn.execute(f"delete from users where id in ({placeholders})", user_ids) for row in image_rows: cleanup_upload_if_unused(conn, row["image_url"] or "") def today_key(): return time.strftime("%Y-%m-%d", time.localtime()) def generation_stats_payload(conn, user_id): today = today_key() rows = conn.execute( """ select kind, sum(count) total_count, sum(case when stat_date = ? then count else 0 end) today_count from generation_stats where user_id = ? group by kind """, (today, user_id), ).fetchall() stats = { "todayOutfitToday": 0, "todayOutfitTotal": 0, "weeklyOutfitToday": 0, "weeklyOutfitTotal": 0, } for row in rows: if row["kind"] == "today": stats["todayOutfitToday"] = row["today_count"] or 0 stats["todayOutfitTotal"] = row["total_count"] or 0 if row["kind"] == "week": stats["weeklyOutfitToday"] = row["today_count"] or 0 stats["weeklyOutfitTotal"] = row["total_count"] or 0 stats["allToday"] = stats["todayOutfitToday"] + stats["weeklyOutfitToday"] stats["allTotal"] = stats["todayOutfitTotal"] + stats["weeklyOutfitTotal"] return stats def record_generation_stat(conn, user_id, kind): if kind not in {"today", "week"}: raise ValueError("不支持的生成类型") now = int(time.time()) conn.execute( """ insert into generation_stats (user_id, stat_date, kind, count, updated_at) values (?, ?, ?, 1, ?) on conflict(user_id, stat_date, kind) do update set count = count + 1, updated_at = excluded.updated_at """, (user_id, today_key(), kind, now), ) def yesterday_key(): return time.strftime("%Y-%m-%d", time.localtime(time.time() - 86400)) def record_user_activity(conn, user_id): now = int(time.time()) date_key = today_key() row = conn.execute( "select active_seconds, last_seen from user_activity_stats where user_id = ? and stat_date = ?", (user_id, date_key), ).fetchone() if row: delta = max(0, min(60, now - int(row["last_seen"] or now))) conn.execute( """ update user_activity_stats set active_seconds = active_seconds + ?, last_seen = ?, updated_at = ? where user_id = ? and stat_date = ? """, (delta, now, now, user_id, date_key), ) else: conn.execute( """ insert into user_activity_stats (user_id, stat_date, active_seconds, last_seen, updated_at) values (?, ?, 0, ?, ?) """, (user_id, date_key, now, now), ) def admin_activity_stats(conn): keys = [today_key(), yesterday_key()] placeholders = ",".join("?" for _ in keys) rows = conn.execute( f""" select stat_date, count(distinct user_id) active_users, avg(active_seconds) avg_seconds from user_activity_stats where stat_date in ({placeholders}) group by stat_date """, keys, ).fetchall() by_date = {row["stat_date"]: row for row in rows} today = by_date.get(keys[0]) yesterday = by_date.get(keys[1]) return { "todayActiveUsers": int(today["active_users"] if today else 0), "yesterdayActiveUsers": int(yesterday["active_users"] if yesterday else 0), "activeUsersTwoDayTotal": int((today["active_users"] if today else 0) + (yesterday["active_users"] if yesterday else 0)), "todayAvgStaySeconds": int(today["avg_seconds"] if today and today["avg_seconds"] is not None else 0), "yesterdayAvgStaySeconds": int(yesterday["avg_seconds"] if yesterday and yesterday["avg_seconds"] is not None else 0), "onlineUsers": int(conn.execute( "select count(distinct user_id) c from user_activity_stats where last_seen >= ?", (int(time.time()) - 120,), ).fetchone()["c"]), } def is_male_gender(gender): return "女" not in str(gender or "") def is_skirt_category_name(category): return str(category or "") in {"裙子", "半身裙", "短裙", "连衣裙"} def filter_wardrobe_by_gender(items, gender): if not is_male_gender(gender): return items return [item for item in items if not is_skirt_category_name(normalize_cloth_category(dict(item)).get("category"))] def normalize_compare_text(value): return "".join(str(value or "").strip().lower().split()) def find_duplicate_wardrobe_item(conn, user_id, item): normalized_item = normalize_cloth_category(dict(item or {})) rows = conn.execute( "select * from wardrobe_items where user_id = ? order by id desc", (user_id,), ).fetchall() target_category = normalize_compare_text(normalized_item.get("category")) target_name = normalize_compare_text(normalized_item.get("name")) target_fields = { key: normalize_compare_text(normalized_item.get(key)) for key in ("color", "style", "fit", "season") } for row in rows: existing = item_payload(row) if normalize_compare_text(existing.get("category")) != target_category: continue if target_name and normalize_compare_text(existing.get("name")) == target_name: return existing match_count = sum( 1 for key, value in target_fields.items() if value and value == normalize_compare_text(existing.get(key)) ) if match_count >= 3: return existing return None def auth_user_id(handler): header = handler.headers.get("Authorization", "") token = header.replace("Bearer ", "").strip() if not token: return None with db() as conn: row = conn.execute( """ select sessions.user_id from sessions join users on users.id = sessions.user_id where sessions.token = ? and coalesce(users.is_blocked, 0) = 0 """, (token,), ).fetchone() return row["user_id"] if row else None def hash_password(password): return hashlib.sha256(password.encode("utf-8")).hexdigest() def default_male_avatar_url(): return configured_value( "DPSTAR_DEFAULT_MALE_AVATAR_URL", configured_value("DPSTAR_DEFAULT_AVATAR_URL", "assets/dpstar-avatar.png"), ) def default_female_avatar_url(): return configured_value("DPSTAR_DEFAULT_FEMALE_AVATAR_URL", default_male_avatar_url()) def default_avatar_for_gender(gender): return default_female_avatar_url() if str(gender or "").strip().startswith("女") else default_male_avatar_url() def switched_default_avatar(current_avatar, gender): current = str(current_avatar or "").strip() known_defaults = { "", "assets/dpstar-avatar.png", configured_value("DPSTAR_DEFAULT_AVATAR_URL", "assets/dpstar-avatar.png"), default_male_avatar_url(), default_female_avatar_url(), } return default_avatar_for_gender(gender) if current in known_defaults else current def public_user(row): return { "id": row["id"], "phone": row["phone"] or "", "nickname": row["nickname"] or f"用户{(row['phone'] or '')[-4:]}", "avatarUrl": row["avatar_url"] or default_avatar_for_gender(row["gender"]), } def create_session(conn, user_id): token = secrets.token_urlsafe(32) conn.execute("insert into sessions (token, user_id, created_at) values (?, ?, ?)", (token, user_id, int(time.time()))) return token def ensure_user_by_phone(conn, phone, password=None): now = int(time.time()) user = conn.execute("select * from users where phone = ?", (phone,)).fetchone() if user: if user["is_blocked"]: raise PermissionError("账号已被拉黑，请联系管理员") if password: conn.execute("update users set password_hash = ?, updated_at = ? where id = ?", (hash_password(password), now, user["id"])) user = conn.execute("select * from users where id = ?", (user["id"],)).fetchone() return user password_hash = hash_password(password) if password else "" cur = conn.execute( """ insert into users (phone, nickname, avatar_url, gender, accessory_mode, password_hash, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?) """, (phone, f"用户{phone[-4:]}", default_male_avatar_url(), "男生", "不需要配饰", password_hash, now, now), ) return conn.execute("select * from users where id = ?", (cur.lastrowid,)).fetchone() def sms_settings_payload(row): return { "enabled": bool(row["enabled"]) if row else True, "yunpianApiKey": row["yunpian_api_key"] or "", "smsSignature": row["sms_signature"] or "", "smsTemplate": row["sms_template"] or "", "updatedAt": row["updated_at"] or 0, } def mask_secret(value): if not value: return "未配置" if len(value) <= 10: return f"{value[:2]}***{value[-2:]}" return f"{value[:6]}***{value[-4:]}" def infer_provider(api_url, model): text = f"{api_url or ''} {model or ''}".lower() if "deepseek" in text: return "DeepSeek" if "dashscope" in text or "qwen" in text or "aliyun" in text: return "通义千问" if "volcengine" in text or "doubao" in text: return "豆包" if "baidu" in text or "ernie" in text: return "文心一言" if "moonshot" in text or "kimi" in text: return "Kimi" if "openai" in text: return "OpenAI" if "zhipu" in text or "glm" in text: return "智谱" return "自定义接口" if api_url else "未接入" def normalize_vision_api_url(api_url, model): value = (api_url or "").strip() lowered = value.lower() model_text = (model or "").lower() if not value or value == "OpenAI Responses / Chat Completions": if "qwen" in model_text or "千问" in model_text: return "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions" return "https://api.openai.com/v1/responses" if lowered == "openai responses / chat completions": return "https://api.openai.com/v1/responses" return value def configured_value(name, default=""): file_first_keys = {"DPSTAR_VISION_ENABLED", "DPSTAR_SAFETY_ENABLED"} start_file = ROOT / "start-web.bat" if name in file_first_keys and start_file.exists(): for line in start_file.read_text(encoding="utf-8", errors="ignore").splitlines(): prefix = f"set {name}=" if line.strip().lower().startswith(prefix.lower()): return line.strip()[len(prefix):].strip() value = os.environ.get(name) if value: return value if start_file.exists(): for line in start_file.read_text(encoding="utf-8", errors="ignore").splitlines(): prefix = f"set {name}=" if line.strip().lower().startswith(prefix.lower()): return line.strip()[len(prefix):].strip() return default def save_start_web_settings(values): start_file = ROOT / "start-web.bat" existing = [] if start_file.exists(): existing = start_file.read_text(encoding="utf-8", errors="ignore").splitlines() if not existing: existing = [ "@echo off", 'cd /d "%~dp0"', f'"{os.environ.get("PYTHON_EXE", "python")}" backend\\server.py', ] set_values = {key: str(value).strip() for key, value in values.items() if value is not None} output = [] handled = set() for line in existing: stripped = line.strip() matched_key = None for key in set_values: if stripped.lower().startswith(f"set {key}=".lower()): matched_key = key break if matched_key: output.append(f"set {matched_key}={set_values[matched_key]}") handled.add(matched_key) else: output.append(line) insert_at = 2 if len(output) >= 2 else len(output) for key, value in set_values.items(): if key not in handled: output.insert(insert_at, f"set {key}={value}") insert_at += 1 start_file.write_text("\n".join(output) + "\n", encoding="utf-8") env_file = Path(os.environ.get("DPSTAR_ENV_FILE", str(DEPLOY_ROOT / "config" / ".env"))).expanduser() if env_file.exists() or os.environ.get("DPSTAR_ENV_FILE"): env_file.parent.mkdir(parents=True, exist_ok=True) env_lines = env_file.read_text(encoding="utf-8", errors="ignore").splitlines() if env_file.exists() else [] env_output = [] env_handled = set() for line in env_lines: stripped = line.strip() matched_key = next((key for key in set_values if stripped.startswith(f"{key}=")), None) if matched_key: env_output.append(f"{matched_key}={set_values[matched_key]}") env_handled.add(matched_key) else: env_output.append(line) for key, value in set_values.items(): if key not in env_handled: env_output.append(f"{key}={value}") env_file.write_text("\n".join(env_output) + "\n", encoding="utf-8") for key, value in set_values.items(): os.environ[key] = value def system_settings_payload(): configured_host = configured_value("DPSTAR_HOST", HOST) configured_port = int(configured_value("DPSTAR_PORT", str(PORT)) or PORT) configured_base_url = configured_value("DPSTAR_BASE_URL", f"http://{configured_host}:{configured_port}") stylist_api_url = configured_value("DPSTAR_STYLIST_API_URL") stylist_model = configured_value("DPSTAR_STYLIST_MODEL", "deepseek-chat") stylist_api_key = configured_value("DPSTAR_STYLIST_API_KEY") vision_model = configured_value("DPSTAR_VISION_MODEL", configured_value("OPENAI_MODEL", "gpt-4.1-mini")) vision_api_url = normalize_vision_api_url(configured_value("DPSTAR_VISION_API_URL", ""), vision_model) vision_api_key = configured_value("DPSTAR_VISION_API_KEY", configured_value("OPENAI_API_KEY")) vision_enabled = configured_value("DPSTAR_VISION_ENABLED", "1") != "0" safety_enabled = configured_value("DPSTAR_SAFETY_ENABLED", "1") != "0" moderation_api_url = configured_value("DPSTAR_IMAGE_MODERATION_API_URL") moderation_model = configured_value("DPSTAR_IMAGE_MODERATION_MODEL", "qwen-vl-plus") moderation_api_key = configured_value("DPSTAR_IMAGE_MODERATION_API_KEY") storage = oss_settings() default_male_avatar = default_male_avatar_url() default_female_avatar = default_female_avatar_url() return { "server": { "host": configured_host, "port": configured_port, "baseUrl": configured_base_url, "runningHost": HOST, "runningPort": PORT, }, "stylist": { "provider": infer_provider(stylist_api_url, stylist_model), "apiUrl": stylist_api_url or "未配置", "model": stylist_model, "apiKey": mask_secret(stylist_api_key), "enabled": bool(stylist_api_url and stylist_api_key), }, "vision": { "provider": infer_provider(vision_api_url, vision_model) if vision_enabled and vision_api_key else "未接入", "apiUrl": vision_api_url, "model": vision_model, "apiKey": mask_secret(vision_api_key), "enabled": bool(vision_enabled and vision_api_key), "enabledByAdmin": vision_enabled, }, "safety": { "enabled": safety_enabled, "provider": infer_provider(moderation_api_url, moderation_model) if safety_enabled and moderation_api_key else "未接入", "apiUrl": moderation_api_url or "未配置", "model": moderation_model or "未配置", "apiKey": mask_secret(moderation_api_key), "textModeration": safety_enabled, "imageModeration": bool(safety_enabled and moderation_api_key and moderation_api_url), }, "storage": { "mode": storage_mode(), "endpoint": storage["endpoint"], "bucket": storage["bucket"], "accessKeyId": mask_secret(storage["access_key_id"]), "accessKeySecret": mask_secret(storage["access_key_secret"]), "prefix": storage["prefix"], "publicBaseUrl": storage["public_base_url"], "ready": all(storage[key] for key in ("endpoint", "bucket", "access_key_id", "access_key_secret")), }, "user": { "defaultAvatarUrl": default_male_avatar, "defaultMaleAvatarUrl": default_male_avatar, "defaultFemaleAvatarUrl": default_female_avatar, }, } def send_yunpian_sms(phone, code, settings): api_key = settings["yunpian_api_key"] or "" signature = settings["sms_signature"] or "DPSTAR AI" template = settings["sms_template"] or "【#signature#】您的验证码是#code#，5分钟内有效。" text = template.replace("#signature#", signature).replace("#code#", code) if not api_key: return {"sent": False, "devCode": code, "message": "未配置云片网 API Key，已生成本地测试验证码"} data = urllib.parse.urlencode({"apikey": api_key, "mobile": phone, "text": text}).encode("utf-8") request = urllib.request.Request( "https://sms.yunpian.com/v2/sms/single_send.json", data=data, headers={"Content-Type": "application/x-www-form-urlencoded;charset=utf-8"}, method="POST", ) with urllib.request.urlopen(request, timeout=30) as response: result = json.loads(response.read().decode("utf-8")) if result.get("code") not in (0, "0"): raise RuntimeError(result.get("msg") or "短信发送失败") return {"sent": True, "message": "验证码已发送"} def create_sms_code(conn, phone): code = f"{secrets.randbelow(900000) + 100000}" now = int(time.time()) settings = conn.execute("select * from sms_settings where id = 1").fetchone() conn.execute( "insert or replace into sms_codes (phone, code, expires_at, created_at) values (?, ?, ?, ?)", (phone, code, now + 300, now), ) return code, settings def verify_sms_code(conn, phone, code): now = int(time.time()) row = conn.execute("select * from sms_codes where phone = ?", (phone,)).fetchone() return bool(row and row["code"] == code and row["expires_at"] >= now) def admin_id_from_token(handler): header = handler.headers.get("Authorization", "") token = header.replace("Bearer ", "").strip() if not token: return None with db() as conn: row = conn.execute("select admin_id from admin_sessions where token = ?", (token,)).fetchone() return row["admin_id"] if row else None def is_admin(handler): return bool(admin_id_from_token(handler)) def item_payload(row): payload = { "id": row["id"], "name": row["name"], "category": row["category"], "color": row["color"] or "", "style": row["style"] or "", "fit": row["fit"] or "", "season": row["season"] or "", "scene": row["scene"] or "", "difficulty": row["difficulty"] or "需要搭配技巧", "image": row["image_url"] or "", "usage": row["usage"] or 0, } return normalize_cloth_fields(normalize_cloth_category(payload)) def has_mojibake(value): return bool(re.search(r"[鑿滈闅澶辫触鐢绌鏃涓鍦鎿]", str(value or ""))) def clean_text_field(value, fallback=""): text = str(value or "").strip() return fallback if not text or has_mojibake(text) else text def normalize_cloth_fields(item): if not item: return item item["name"] = clean_text_field(item.get("name"), "待确认单品") item["color"] = clean_text_field(item.get("color"), "基础色") item["style"] = clean_text_field(item.get("style"), "简约") item["fit"] = clean_text_field(item.get("fit"), "合身") item["season"] = clean_text_field(item.get("season"), "四季") item["scene"] = clean_text_field(item.get("scene"), "日常") item["difficulty"] = clean_text_field(item.get("difficulty"), "百搭") return item def normalize_cloth_category(item): if not item: return item text = f"{item.get('category', '')} {item.get('name', '')}".lower() rules = [ ("外套", ["外套", "夹克", "西装", "风衣", "大衣", "羽绒", "开衫", "马甲", "jacket", "coat", "blazer"]), ("鞋子", ["鞋子", "鞋", "板鞋", "球鞋", "运动鞋", "乐福", "sneaker", "shoe", "boot"]), ("配饰", ["配饰", "包", "帽", "手表", "项链", "戒指", "腰带", "围巾", "眼镜", "bag", "watch", "cap", "hat"]), ("连衣裙", ["连衣裙", "长裙", "one piece", "dress"]), ("半身裙", ["半身裙", "半裙", "skirt"]), ("短裙", ["短裙", "mini skirt"]), ("裤子", ["裤子", "短裤", "长裤", "牛仔裤", "休闲裤", "工装裤", "西裤", "下装", "pants", "jeans", "shorts", "trousers"]), ("上衣", ["上衣", "t恤", "t-shirt", "衬衫", "卫衣", "毛衣", "针织", "背心", "polo", "shirt", "hoodie", "sweater", "top"]), ] if item.get("category") in {"上衣", "裤子", "半身裙", "短裙", "连衣裙", "外套", "鞋子", "配饰"}: return item for category, words in rules: if any(word in text for word in words): item["category"] = category return item item["category"] = "上衣" return item def recognition_image_data(image_reference): local_path = upload_file_path(image_reference) if local_path: return local_path.read_bytes(), mimetypes.guess_type(local_path.name)[0] or "image/jpeg" object_key = oss_object_key_from_url(image_reference) if object_key: image_url = oss_object_url(oss_settings(), object_key) request = urllib.request.Request(image_url, headers={"User-Agent": "DPSTAR/1.0"}) with urllib.request.urlopen(request, timeout=60) as response: data = response.read(MAX_UPLOAD_BYTES + 1) content_type = response.headers.get_content_type() or mimetypes.guess_type(object_key)[0] or "image/jpeg" if len(data) > MAX_UPLOAD_BYTES: raise ValueError("OSS 图片超过 8MB，无法识别") if not data: raise ValueError("OSS 图片内容为空") return data, content_type raise ValueError("找不到待识别图片，请重新上传") def call_openai_vision(image_reference): if configured_value("DPSTAR_VISION_ENABLED", "1") == "0": return None api_key = configured_value("DPSTAR_VISION_API_KEY", configured_value("OPENAI_API_KEY")) model = configured_value("DPSTAR_VISION_MODEL", configured_value("OPENAI_MODEL", "gpt-4.1-mini")) api_url = normalize_vision_api_url(configured_value("DPSTAR_VISION_API_URL", ""), model) if not api_key: return None image_bytes, content_type = recognition_image_data(image_reference) data_url = f"data:{content_type};base64," + base64.b64encode(image_bytes).decode("ascii") prompt = "识别这件衣服，返回 JSON：name, category, color, style, fit, season, scene, difficulty。category 只能从上衣、裤子、半身裙、短裙、连衣裙、外套、鞋子、配饰中选择；半身裙、短裙、连衣裙要区分准确。不要输出多余文字。" if "responses" in api_url: payload = { "model": model, "input": [ { "role": "user", "content": [ {"type": "input_text", "text": prompt}, {"type": "input_image", "image_url": data_url}, ], } ], } else: payload = { "model": model, "messages": [ { "role": "user", "content": [ {"type": "text", "text": prompt}, {"type": "image_url", "image_url": {"url": data_url}}, ], } ], "temperature": 0.2, } request = urllib.request.Request( api_url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers={ "Content-Type": "application/json", "Authorization": f"Bearer {api_key}", }, method="POST", ) with urllib.request.urlopen(request, timeout=60) as response: data = json.loads(response.read().decode("utf-8")) text = data.get("output_text", "") if not text: choices = data.get("choices", []) for choice in choices: content = choice.get("message", {}).get("content", "") if isinstance(content, str): text += content elif isinstance(content, list): text += "".join(part.get("text", "") for part in content if isinstance(part, dict)) if not text: parts = data.get("output", []) for part in parts: for content in part.get("content", []): if content.get("type") == "output_text": text += content.get("text", "") if text.startswith("```"): text = text.strip("`").replace("json\n", "", 1).replace("JSON\n", "", 1).strip() if not text.startswith("{"): start = text.find("{") end = text.rfind("}") if start >= 0 and end > start: text = text[start : end + 1] try: return normalize_cloth_category(json.loads(text)) except json.JSONDecodeError: return {"name": "图片识别单品", "category": "上衣", "color": "", "style": "", "fit": "", "season": "", "scene": "", "difficulty": "需要搭配技巧"} def mock_ai_result(image_url): return normalize_cloth_category({ "name": "待确认单品", "category": "上衣", "color": "黑白", "style": "简约", "fit": "合身", "season": "四季", "scene": "日常、通勤", "difficulty": "百搭", "image": image_url, "note": "当前未配置衣服图片识别 API Key，返回本地示例识别结果。", }) def user_state_from_payload(payload): city = payload.get("profileCity") or payload.get("todayCity") or "" return { "gender": payload.get("gender", "男生"), "height": payload.get("height", ""), "weight": payload.get("weight", ""), "age_info": payload.get("ageInfo", ""), "size": payload.get("size", ""), "preferred_style": payload.get("preferredStyle", ""), "common_scenes": payload.get("commonScenes", ""), "profile_city": city, "problem": payload.get("problem", ""), "today_city": city, "today_scene": payload.get("todayScene", ""), "today_mood": payload.get("todayMood", ""), "accessory_mode": payload.get("accessoryMode", "不需要配饰"), } def user_payload(row): return { "id": row["id"], "phone": row["phone"] or "", "nickname": row["nickname"] or "", "avatarUrl": row["avatar_url"] or default_avatar_for_gender(row["gender"]), "gender": row["gender"] or "男生", "height": row["height"] or "", "weight": row["weight"] or "", "ageInfo": row["age_info"] or "", "size": row["size"] or "", "preferredStyle": row["preferred_style"] or "", "commonScenes": row["common_scenes"] or "", "profileCity": row["profile_city"] or "", "problem": row["problem"] or "", "todayCity": row["today_city"] or "", "todayScene": row["today_scene"] or "", "todayMood": row["today_mood"] or "", "accessoryMode": row["accessory_mode"] or "不需要配饰", "wardrobeLimit": row["wardrobe_limit"] or 50, "hasPassword": bool(row["password_hash"]), "isBlocked": bool(row["is_blocked"]), "createdAt": row["created_at"], "updatedAt": row["updated_at"] or row["created_at"], } def local_outfit_recommendation(payload): profile = payload.get("profile") or {} wardrobe = filter_wardrobe_by_gender(payload.get("wardrobe") or [], profile.get("gender", "男生")) seed = int(payload.get("seed") or 0) scene = payload.get("scene") or "日常" mood = payload.get("mood") or "干净" city = payload.get("city") or "城市未填写" needs_accessory = payload.get("accessoryMode") != "不需要配饰" def pick(categories): options = [item for item in wardrobe if item.get("category") in categories] if not options: category = categories[0] return { "name": f"待补充{category}", "category": category, "color": "基础色", "style": "简约", "fit": "合身", "image": "assets/white-tshirt.png", } versatile = [item for item in options if item.get("difficulty") == "百搭"] pool = versatile or options return pool[seed % len(pool)] bottom_categories = ["裤子"] if is_male_gender(profile.get("gender")) else ["裤子", "半身裙", "短裙", "连衣裙", "裙子"] bottom = pick(bottom_categories) dress_text = f"{bottom.get('name', '')}{bottom.get('style', '')}{bottom.get('fit', '')}".lower() use_dress_as_main = bottom.get("category") == "连衣裙" or any( word in dress_text for word in ("连衣裙", "长裙", "one piece", "dress") ) top = None if use_dress_as_main else pick(["上衣"]) shoes = pick(["鞋子"]) accessory = pick(["配饰", "外套"]) items = [ *([] if use_dress_as_main else [{"role": "上衣", "product": top, "reason": f"{top.get('color', '基础色')}清爽好搭，适合{mood}需求。"}]), {"role": "连衣裙" if use_dress_as_main else "下装", "product": bottom, "reason": "连衣裙可直接作为主体穿搭，不需要额外搭配上衣。" if use_dress_as_main else f"{bottom.get('fit', '合身')}版型更利落，和上衣比例协调。"}, {"role": "鞋子", "product": shoes, "reason": f"{shoes.get('color', '基础色')}和整体呼应，提升完成度。"}, ] if needs_accessory: items.append({"role": "配饰", "product": accessory, "reason": "增加细节，但不抢整体干净感。"}) body_parts = [] if profile.get("ageInfo"): body_parts.append(f"年龄{profile.get('ageInfo')}") if profile.get("height"): body_parts.append(f"身高{profile.get('height')}") if profile.get("weight"): body_parts.append(f"体重{profile.get('weight')}") body_text = "、".join(body_parts) body_analysis = f"，结合{body_text}，建议注意比例和利落度" if body_text else "" notice_text = "根据当天实际体感选择厚薄；如果出门时间较长，可以预留一件轻便外套。" return { "analysis": f"根据{city}的城市天气和体感特点，今天先以舒适、干净、不厚重为主；出门场景是{scene}，想要感觉是{mood}{body_analysis}。", "style": "干净少年感休闲风" if "约会" in scene else "干净日常", "highlight": f"这套优先使用衣橱已有单品，颜色控制在3个主色内，适合{scene}。", "notice": notice_text, "items": items, } def call_ai_stylist_api(payload): """ 鍦ㄨ繖閲屾帴鍏ヤ綘鐨?AI 鎼厤甯?API銆? 瑕佹眰杩斿洖缁撴瀯锛? { "analysis": "鍏堝垎鏋愬煄甯傘€佸満鏅€佺敤鎴疯祫鏂欏拰琛ｆ┍", "style": "鏁翠綋椋庢牸", "highlight": "鎼厤浜偣", "notice": "娉ㄦ剰浜嬮」", "items": [ {"role": "涓婅。", "product": {"name": "...", "category": "...", "image": "..."}, "reason": "..."} ] } """ api_key = configured_value("DPSTAR_STYLIST_API_KEY") api_url = configured_value("DPSTAR_STYLIST_API_URL") model = configured_value("DPSTAR_STYLIST_MODEL", "deepseek-chat") if not api_key or not api_url: return None profile = payload.get("profile") or {} safe_payload = dict(payload) safe_payload["wardrobe"] = filter_wardrobe_by_gender(payload.get("wardrobe") or [], profile.get("gender", "男生")) prompt = { "role": "system", "content": ( "你是 DPSTAR AI 的 AI衣橱搭配师。请先根据城市、出门场景、用户想要的感觉，以及用户填写的年龄、身高、体重进行分析；如果年龄、身高或体重为空，就不要提及该项。" "然后优先从已有衣橱中推荐最合理的穿搭。必须先分析，再推荐，不能随意推荐。只返回 JSON，不要 Markdown，不要额外解释。" "如果用户是男生，衣橱和推荐里都不能出现裙子、半身裙、短裙、连衣裙，analysis 里也不要建议裙装。" "如果用户是女生且推荐裙装，请区分半身裙、短裙、连衣裙：半身裙和短裙需要搭配上衣；连衣裙可作为主体，不要再推荐上衣，除非需要外套。" "JSON 格式包含：analysis, style, highlight, notice, items。" "items 数组每项包含 role, product, reason；product 尽量使用用户衣橱里的完整单品对象。" ), } body = { "model": model, "temperature": 0.7, "messages": [ prompt, { "role": "user", "content": json.dumps(safe_payload, ensure_ascii=False), }, ], } request = urllib.request.Request( api_url, data=json.dumps(body, ensure_ascii=False).encode("utf-8"), headers={ "Content-Type": "application/json", "Authorization": f"Bearer {api_key}", }, method="POST", ) with urllib.request.urlopen(request, timeout=60) as response: data = json.loads(response.read().decode("utf-8")) content = ( data.get("choices", [{}])[0] .get("message", {}) .get("content", "") .strip() ) if content.startswith("```"): content = content.strip("`") content = content.replace("json\n", "", 1).replace("JSON\n", "", 1).strip() if not content.startswith("{"): start = content.find("{") end = content.rfind("}") if start >= 0 and end > start: content = content[start : end + 1] result = json.loads(content) if is_male_gender(profile.get("gender")): items = result.get("items") if isinstance(result, dict) else [] if any(is_skirt_category_name(normalize_cloth_category(dict((item or {}).get("product") or item or {})).get("category")) for item in (items or [])): return None return result class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(ROOT), **kwargs) def translate_path(self, path): request_path = urllib.parse.urlparse(path).path if request_path.startswith("/uploads/"): filename = Path(request_path).name return str((UPLOAD_DIR / filename).resolve()) return super().translate_path(path) def do_OPTIONS(self): json_response(self, {"ok": True}) def do_POST(self): if self.path == "/api/admin/login": payload = read_json(self) username = (payload.get("username") or "").strip() password = payload.get("password") or "" with db() as conn: admin = conn.execute("select * from admins where username = ?", (username,)).fetchone() if not admin or admin["password_hash"] != hash_password(password): json_response(self, {"error": "管理员账号或密码错误"}, HTTPStatus.UNAUTHORIZED) return token = secrets.token_urlsafe(32) conn.execute( "insert into admin_sessions (token, admin_id, created_at) values (?, ?, ?)", (token, admin["id"], int(time.time())), ) json_response(self, {"token": token, "admin": {"id": admin["id"], "username": admin["username"], "permissions": admin["permissions"] or "all"}}) return if self.path == "/api/admin/create": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) username = (payload.get("username") or "").strip() password = payload.get("password") or "" permissions = (payload.get("permissions") or "all").strip() if permissions not in {"all", "users", "settings", "readonly"}: permissions = "all" if len(username) < 3 or len(password) < 6: json_response(self, {"error": "账号至少3位，密码至少6位"}, HTTPStatus.BAD_REQUEST) return now = int(time.time()) try: with db() as conn: conn.execute( "insert into admins (username, password_hash, permissions, created_at, updated_at) values (?, ?, ?, ?, ?)", (username, hash_password(password), permissions, now, now), ) except sqlite3.IntegrityError: json_response(self, {"error": "管理员账号已存在"}, HTTPStatus.BAD_REQUEST) return json_response(self, {"ok": True}) return if self.path == "/api/admin/password": admin_id = admin_id_from_token(self) if not admin_id: json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) old_password = payload.get("oldPassword") or "" new_password = payload.get("newPassword") or "" if len(new_password) < 6: json_response(self, {"error": "新密码至少6位"}, HTTPStatus.BAD_REQUEST) return with db() as conn: admin = conn.execute("select * from admins where id = ?", (admin_id,)).fetchone() if not admin or admin["password_hash"] != hash_password(old_password): json_response(self, {"error": "原密码错误"}, HTTPStatus.BAD_REQUEST) return conn.execute( "update admins set password_hash = ?, updated_at = ? where id = ?", (hash_password(new_password), int(time.time()), admin_id), ) json_response(self, {"ok": True}) return if self.path == "/api/admin/sms-settings": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) with db() as conn: conn.execute( """ update sms_settings set enabled = ?, yunpian_api_key = ?, sms_signature = ?, sms_template = ?, updated_at = ? where id = 1 """, ( 0 if str(payload.get("enabled")) == "0" else 1, payload.get("yunpianApiKey", "").strip(), payload.get("smsSignature", "").strip(), payload.get("smsTemplate", "").strip(), int(time.time()), ), ) json_response(self, {"ok": True}) return if self.path == "/api/admin/system-settings": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) if reject_sensitive_payload(self, {key: value for key, value in payload.items() if "key" not in key.lower()}): return host = (payload.get("host") or "127.0.0.1").strip() port_text = str(payload.get("port") or "8000").strip() if not port_text.isdigit() or not (1 <= int(port_text) <= 65535): json_response(self, {"error": "端口必须是 1-65535 的数字"}, HTTPStatus.BAD_REQUEST) return previous_default_avatars = { "", "assets/dpstar-avatar.png", configured_value("DPSTAR_DEFAULT_AVATAR_URL", "assets/dpstar-avatar.png"), default_male_avatar_url(), default_female_avatar_url(), } values = { "DPSTAR_HOST": host, "DPSTAR_PORT": port_text, "DPSTAR_BASE_URL": (payload.get("baseUrl") or f"http://{host}:{port_text}").strip(), "DPSTAR_STYLIST_API_URL": (payload.get("stylistApiUrl") or "").strip(), "DPSTAR_STYLIST_MODEL": (payload.get("stylistModel") or "deepseek-chat").strip(), "DPSTAR_SAFETY_ENABLED": "0" if str(payload.get("safetyEnabled")) == "0" else "1", "DPSTAR_IMAGE_MODERATION_API_URL": (payload.get("moderationApiUrl") or "").strip(), "DPSTAR_IMAGE_MODERATION_MODEL": (payload.get("moderationModel") or "").strip(), "DPSTAR_STORAGE_MODE": "oss" if str(payload.get("storageMode")).lower() == "oss" else "local", "DPSTAR_OSS_ENDPOINT": (payload.get("ossEndpoint") or configured_value("DPSTAR_OSS_ENDPOINT", "")).strip(), "DPSTAR_OSS_BUCKET": (payload.get("ossBucket") or configured_value("DPSTAR_OSS_BUCKET", "")).strip(), "DPSTAR_OSS_PREFIX": (payload.get("ossPrefix") or configured_value("DPSTAR_OSS_PREFIX", "uploads") or "uploads").strip(), "DPSTAR_OSS_PUBLIC_BASE_URL": (payload.get("ossPublicBaseUrl") or configured_value("DPSTAR_OSS_PUBLIC_BASE_URL", "")).strip(), "DPSTAR_DEFAULT_AVATAR_URL": (payload.get("defaultMaleAvatarUrl") or payload.get("defaultAvatarUrl") or "assets/dpstar-avatar.png").strip(), "DPSTAR_DEFAULT_MALE_AVATAR_URL": (payload.get("defaultMaleAvatarUrl") or payload.get("defaultAvatarUrl") or "assets/dpstar-avatar.png").strip(), "DPSTAR_DEFAULT_FEMALE_AVATAR_URL": (payload.get("defaultFemaleAvatarUrl") or payload.get("defaultMaleAvatarUrl") or payload.get("defaultAvatarUrl") or "assets/dpstar-avatar.png").strip(), "DPSTAR_VISION_ENABLED": "0" if str(payload.get("visionEnabled")) == "0" else "1", "DPSTAR_VISION_MODEL": (payload.get("visionModel") or "gpt-4.1-mini").strip(), "OPENAI_MODEL": (payload.get("visionModel") or "gpt-4.1-mini").strip(), } values["DPSTAR_VISION_API_URL"] = normalize_vision_api_url(payload.get("visionApiUrl"), values["DPSTAR_VISION_MODEL"]) stylist_key = (payload.get("stylistApiKey") or "").strip() vision_key = (payload.get("visionApiKey") or "").strip() if stylist_key: values["DPSTAR_STYLIST_API_KEY"] = stylist_key if vision_key: values["DPSTAR_VISION_API_KEY"] = vision_key values["OPENAI_API_KEY"] = vision_key moderation_key = (payload.get("moderationApiKey") or "").strip() if moderation_key: values["DPSTAR_IMAGE_MODERATION_API_KEY"] = moderation_key oss_access_key_id = (payload.get("ossAccessKeyId") or "").strip() oss_access_key_secret = (payload.get("ossAccessKeySecret") or "").strip() if oss_access_key_id: values["DPSTAR_OSS_ACCESS_KEY_ID"] = oss_access_key_id if oss_access_key_secret: values["DPSTAR_OSS_ACCESS_KEY_SECRET"] = oss_access_key_secret if values["DPSTAR_STORAGE_MODE"] == "oss": effective_access_id = oss_access_key_id or configured_value("DPSTAR_OSS_ACCESS_KEY_ID", "") effective_secret = oss_access_key_secret or configured_value("DPSTAR_OSS_ACCESS_KEY_SECRET", "") if not values["DPSTAR_OSS_ENDPOINT"] or not values["DPSTAR_OSS_BUCKET"] or not effective_access_id or not effective_secret: json_response(self, {"error": "启用 OSS 前请完整填写 Endpoint、Bucket、AccessKey ID 和 Secret"}, HTTPStatus.BAD_REQUEST) return save_start_web_settings(values) with db() as conn: users = conn.execute("select id, gender, avatar_url from users").fetchall() for user in users: if str(user["avatar_url"] or "").strip() in previous_default_avatars: conn.execute( "update users set avatar_url = ?, updated_at = ? where id = ?", (default_avatar_for_gender(user["gender"]), int(time.time()), user["id"]), ) json_response(self, {"ok": True, "settings": system_settings_payload()}) return if self.path == "/api/admin/menu-settings": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) items = payload.get("items", []) if reject_sensitive_payload(self, {"items": [{key: value for key, value in item.items() if key != "iconUrl"} for item in items if isinstance(item, dict)]}): return write_bottom_menu(items) json_response(self, {"ok": True, "bottomMenu": read_bottom_menu()}) return if self.path == "/api/admin/profile-menu-settings": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) items = payload.get("items", []) if reject_sensitive_payload(self, {"items": [{key: value for key, value in item.items() if key != "iconUrl"} for item in items if isinstance(item, dict)]}): return write_profile_menu(items) json_response(self, {"ok": True, "profileMenu": read_profile_menu()}) return if self.path == "/api/admin/action-button-settings": if not is_admin(self): json_response(self, {"error": "无管理员权限"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) items = payload.get("items", []) if reject_sensitive_payload(self, {"items": [{key: value for key, value in item.items() if key != "iconUrl"} for item in items if isinstance(item, dict)]}): return write_action_buttons(items) json_response(self, {"ok": True, "actionButtons": read_action_buttons()}) return if self.path == "/api/admin/privacy-policy": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) title = payload.get("title") or "AI衣橱搭配师隐私政策" content = payload.get("content") or "" if not content.strip(): json_response(self, {"error": "请填写隐私条款内容"}, HTTPStatus.BAD_REQUEST) return write_privacy_policy({"title": title, "content": content}) json_response(self, {"ok": True, "privacyPolicy": read_privacy_policy()}) return if self.path == "/api/admin/privacy-policy/upload": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return form = cgi.FieldStorage(fp=self.rfile, headers=self.headers, environ={"REQUEST_METHOD": "POST"}) field = form["file"] if "file" in form else None if field is None or not field.filename: json_response(self, {"error": "娌℃湁鏀跺埌闅愮鏉℃鏂囦欢"}, HTTPStatus.BAD_REQUEST) return filename = Path(field.filename).name suffix = Path(filename).suffix.lower() raw = field.file.read() if suffix == ".docx": temp_path = DATA_DIR / f"privacy_upload_{secrets.token_hex(8)}.docx" temp_path.write_bytes(raw) content = extract_docx_text(temp_path) temp_path.unlink(missing_ok=True) elif suffix in {".txt", ".md"}: content = raw.decode("utf-8", errors="ignore") else: json_response(self, {"error": "仅支持上传 docx、txt、md 文件"}, HTTPStatus.BAD_REQUEST) return if not content.strip(): json_response(self, {"error": "文件内容为空或无法识别"}, HTTPStatus.BAD_REQUEST) return title = content.splitlines()[0].strip()[:80] or "AI衣橱搭配师隐私政策" write_privacy_policy({"title": title, "content": content}) json_response(self, {"ok": True, "privacyPolicy": read_privacy_policy()}) return if self.path == "/api/admin/user-agreement": if not is_admin(self): json_response(self, {"error": "无管理员权限"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) title = payload.get("title") or "DPSTAR AI 用户协议" content = payload.get("content") or "" if not content.strip(): json_response(self, {"error": "请填写用户协议内容"}, HTTPStatus.BAD_REQUEST) return write_user_agreement({"title": title, "content": content}) json_response(self, {"ok": True, "userAgreement": read_user_agreement()}) return if self.path == "/api/admin/user-agreement/upload": if not is_admin(self): json_response(self, {"error": "无管理员权限"}, HTTPStatus.UNAUTHORIZED) return form = cgi.FieldStorage(fp=self.rfile, headers=self.headers, environ={"REQUEST_METHOD": "POST"}) field = form["file"] if "file" in form else None if field is None or not field.filename: json_response(self, {"error": "没有收到用户协议文件"}, HTTPStatus.BAD_REQUEST) return filename = Path(field.filename).name suffix = Path(filename).suffix.lower() raw = field.file.read() if suffix == ".docx": temp_path = DATA_DIR / f"agreement_upload_{secrets.token_hex(8)}.docx" temp_path.write_bytes(raw) content = extract_docx_text(temp_path) temp_path.unlink(missing_ok=True) elif suffix in {".txt", ".md"}: content = raw.decode("utf-8", errors="ignore") else: json_response(self, {"error": "仅支持上传 docx、txt、md 文件"}, HTTPStatus.BAD_REQUEST) return if not content.strip(): json_response(self, {"error": "文件内容为空或无法识别"}, HTTPStatus.BAD_REQUEST) return title = content.splitlines()[0].strip()[:80] or "DPSTAR AI 用户协议" write_user_agreement({"title": title, "content": content}) json_response(self, {"ok": True, "userAgreement": read_user_agreement()}) return if self.path == "/api/admin/users/batch": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) action = payload.get("action") user_ids = [int(user_id) for user_id in payload.get("userIds", []) if str(user_id).isdigit()] if not user_ids: json_response(self, {"error": "璇峰厛閫夋嫨鐢ㄦ埛"}, HTTPStatus.BAD_REQUEST) return placeholders = ",".join("?" for _ in user_ids) now = int(time.time()) with db() as conn: if action == "block": conn.execute( f"update users set is_blocked = 1, updated_at = ? where id in ({placeholders})", (now, *user_ids), ) conn.execute(f"delete from sessions where user_id in ({placeholders})", user_ids) elif action == "delete": delete_users_and_uploads(conn, user_ids) else: json_response(self, {"error": "涓嶆敮鎸佺殑鎵归噺鎿嶄綔"}, HTTPStatus.BAD_REQUEST) return json_response(self, {"ok": True, "count": len(user_ids)}) return if self.path == "/api/admin/upload": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return form = cgi.FieldStorage(fp=self.rfile, headers=self.headers, environ={"REQUEST_METHOD": "POST"}) field = form["file"] if "file" in form else None if field is None or not field.filename: json_response(self, {"error": "娌℃湁鏀跺埌鍥剧墖"}, HTTPStatus.BAD_REQUEST) return try: url = save_checked_upload(field, "admin") except ValueError as exc: json_response(self, {"error": str(exc)}, HTTPStatus.BAD_REQUEST) return json_response(self, {"url": url}) return if self.path == "/api/auth/register": payload = read_json(self) phone = (payload.get("phone") or "").strip() password = payload.get("password") or "" if not phone or len(phone) != 11 or not phone.isdigit(): json_response(self, {"error": "请输入正确的手机号"}, HTTPStatus.BAD_REQUEST) return if len(password) < 6: json_response(self, {"error": "密码至少6位"}, HTTPStatus.BAD_REQUEST) return with db() as conn: existing = conn.execute("select * from users where phone = ?", (phone,)).fetchone() if existing: json_response(self, {"error": "该手机号已经存在，请直接登录"}, HTTPStatus.CONFLICT) return user = ensure_user_by_phone(conn, phone, password) token = create_session(conn, user["id"]) json_response(self, {"token": token, "user": public_user(user)}) return if self.path == "/api/auth/password-login": payload = read_json(self) phone = (payload.get("phone") or "").strip() password = payload.get("password") or "" with db() as conn: user = conn.execute("select * from users where phone = ?", (phone,)).fetchone() if not user or not user["password_hash"] or user["password_hash"] != hash_password(password): json_response(self, {"error": "手机号或密码错误"}, HTTPStatus.UNAUTHORIZED) return if user["is_blocked"]: json_response(self, {"error": "账号已被拉黑，请联系管理员"}, HTTPStatus.FORBIDDEN) return token = create_session(conn, user["id"]) json_response(self, {"token": token, "user": public_user(user)}) return if self.path == "/api/auth/send-code": payload = read_json(self) phone = (payload.get("phone") or "").strip() if not phone or len(phone) != 11 or not phone.isdigit(): json_response(self, {"error": "请输入正确的手机号"}, HTTPStatus.BAD_REQUEST) return code = f"{secrets.randbelow(900000) + 100000}" now = int(time.time()) with db() as conn: settings = conn.execute("select * from sms_settings where id = 1").fetchone() if settings and not settings["enabled"]: json_response(self, {"error": "短信验证码功能已关闭，请使用密码登录"}, HTTPStatus.SERVICE_UNAVAILABLE) return account_exists = bool(conn.execute("select 1 from users where phone = ?", (phone,)).fetchone()) conn.execute( "insert or replace into sms_codes (phone, code, expires_at, created_at) values (?, ?, ?, ?)", (phone, code, now + 300, now), ) try: result = send_yunpian_sms(phone, code, settings) except Exception as exc: json_response(self, {"error": f"鐭俊鍙戦€佸け璐ワ細{exc}"}, HTTPStatus.BAD_GATEWAY) return result["accountExists"] = account_exists json_response(self, result) return if self.path == "/api/auth/sms-login": payload = read_json(self) phone = (payload.get("phone") or "").strip() code = (payload.get("code") or "").strip() now = int(time.time()) with db() as conn: settings = conn.execute("select * from sms_settings where id = 1").fetchone() if settings and not settings["enabled"]: json_response(self, {"error": "短信验证码功能已关闭，请使用密码登录"}, HTTPStatus.SERVICE_UNAVAILABLE) return row = conn.execute("select * from sms_codes where phone = ?", (phone,)).fetchone() if not row or row["code"] != code or row["expires_at"] < now: json_response(self, {"error": "验证码错误或已过期"}, HTTPStatus.BAD_REQUEST) return try: user = ensure_user_by_phone(conn, phone) except PermissionError as exc: json_response(self, {"error": str(exc)}, HTTPStatus.FORBIDDEN) return conn.execute("delete from sms_codes where phone = ?", (phone,)) token = create_session(conn, user["id"]) json_response(self, {"token": token, "user": public_user(user)}) return if self.path == "/api/account/send-bind-code": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "未登录"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) phone = (payload.get("phone") or "").strip() code_type = payload.get("type") or "new" if not phone or len(phone) != 11 or not phone.isdigit(): json_response(self, {"error": "请输入正确的手机号"}, HTTPStatus.BAD_REQUEST) return with db() as conn: user = conn.execute("select * from users where id = ?", (user_id,)).fetchone() if not user: json_response(self, {"error": "用户不存在"}, HTTPStatus.NOT_FOUND) return if code_type == "old" and (user["phone"] or "") != phone: json_response(self, {"error": "原手机号不匹配"}, HTTPStatus.BAD_REQUEST) return if code_type == "new": existing = conn.execute("select id from users where phone = ? and id != ?", (phone, user_id)).fetchone() if existing: json_response(self, {"error": "璇ユ墜鏈哄彿宸茬粦瀹氬叾浠栬处鍙"}, HTTPStatus.BAD_REQUEST) return code, settings = create_sms_code(conn, phone) try: result = send_yunpian_sms(phone, code, settings) except Exception as exc: json_response(self, {"error": f"鐭俊鍙戦€佸け璐ワ細{exc}"}, HTTPStatus.BAD_GATEWAY) return json_response(self, result) return if self.path == "/api/account/phone": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "未登录"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) new_phone = (payload.get("newPhone") or "").strip() old_code = (payload.get("oldCode") or "").strip() new_code = (payload.get("newCode") or "").strip() if not new_phone or len(new_phone) != 11 or not new_phone.isdigit(): json_response(self, {"error": "璇疯緭鍏ユ纭殑鏂版墜鏈哄彿"}, HTTPStatus.BAD_REQUEST) return with db() as conn: user = conn.execute("select * from users where id = ?", (user_id,)).fetchone() if not user: json_response(self, {"error": "用户不存在"}, HTTPStatus.NOT_FOUND) return old_phone = user["phone"] or "" if old_phone and not verify_sms_code(conn, old_phone, old_code): json_response(self, {"error": "原手机号验证码错误或已过期"}, HTTPStatus.BAD_REQUEST) return if not verify_sms_code(conn, new_phone, new_code): json_response(self, {"error": "新手机号验证码错误或已过期"}, HTTPStatus.BAD_REQUEST) return existing = conn.execute("select id from users where phone = ? and id != ?", (new_phone, user_id)).fetchone() if existing: json_response(self, {"error": "该手机号已绑定其他账号"}, HTTPStatus.BAD_REQUEST) return now = int(time.time()) conn.execute("update users set phone = ?, updated_at = ? where id = ?", (new_phone, now, user_id)) if old_phone: conn.execute("delete from sms_codes where phone = ?", (old_phone,)) conn.execute("delete from sms_codes where phone = ?", (new_phone,)) user = conn.execute("select * from users where id = ?", (user_id,)).fetchone() json_response(self, {"ok": True, "profile": user_payload(user)}) return if self.path == "/api/login": payload = read_json(self) if reject_sensitive_payload(self, {key: value for key, value in payload.items() if key not in ("phone", "avatarUrl")}): return phone = (payload.get("phone") or "").strip() if not phone or len(phone) != 11 or not phone.isdigit(): json_response(self, {"error": "请输入正确的手机号"}, HTTPStatus.BAD_REQUEST) return nickname = payload.get("nickname") or f"鐢ㄦ埛{phone[-4:]}" user_state = user_state_from_payload(payload) avatar_url = payload.get("avatarUrl") or default_avatar_for_gender(user_state["gender"]) now = int(time.time()) token = secrets.token_urlsafe(32) with db() as conn: user = conn.execute("select * from users where phone = ?", (phone,)).fetchone() if user: if user["is_blocked"]: json_response(self, {"error": "账号已被拉黑，请联系管理员"}, HTTPStatus.FORBIDDEN) return user_id = user["id"] conn.execute( """ update users set nickname = ?, avatar_url = ?, gender = coalesce(nullif(?, ''), gender), height = coalesce(nullif(?, ''), height), weight = coalesce(nullif(?, ''), weight), age_info = coalesce(nullif(?, ''), age_info), size = coalesce(nullif(?, ''), size), preferred_style = coalesce(nullif(?, ''), preferred_style), common_scenes = coalesce(nullif(?, ''), common_scenes), profile_city = coalesce(nullif(?, ''), profile_city), problem = coalesce(nullif(?, ''), problem), today_city = coalesce(nullif(?, ''), today_city), today_scene = coalesce(nullif(?, ''), today_scene), today_mood = coalesce(nullif(?, ''), today_mood), accessory_mode = coalesce(nullif(?, ''), accessory_mode), updated_at = ? where id = ? """, ( nickname, avatar_url, user_state["gender"], user_state["height"], user_state["weight"], user_state["age_info"], user_state["size"], user_state["preferred_style"], user_state["common_scenes"], user_state["profile_city"], user_state["problem"], user_state["today_city"], user_state["today_scene"], user_state["today_mood"], user_state["accessory_mode"], now, user_id, ), ) else: cur = conn.execute( """ insert into users (phone, nickname, avatar_url, gender, height, weight, age_info, size, preferred_style, common_scenes, profile_city, problem, today_city, today_scene, today_mood, accessory_mode, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( phone, nickname, avatar_url, user_state["gender"], user_state["height"], user_state["weight"], user_state["age_info"], user_state["size"], user_state["preferred_style"], user_state["common_scenes"], user_state["profile_city"], user_state["problem"], user_state["today_city"], user_state["today_scene"], user_state["today_mood"], user_state["accessory_mode"], now, now, ), ) user_id = cur.lastrowid conn.execute("insert into sessions (token, user_id, created_at) values (?, ?, ?)", (token, user_id, now)) json_response(self, {"token": token, "user": {"id": user_id, "phone": phone, "nickname": nickname, "avatarUrl": avatar_url}}) return if self.path == "/api/profile": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) if reject_sensitive_payload(self, payload): return user_state = user_state_from_payload(payload) now = int(time.time()) with db() as conn: current_user = conn.execute("select avatar_url from users where id = ?", (user_id,)).fetchone() avatar_url = switched_default_avatar(current_user["avatar_url"] if current_user else "", user_state["gender"]) conn.execute( """ update users set avatar_url = ?, gender = ?, height = ?, weight = ?, age_info = ?, size = ?, preferred_style = ?, common_scenes = ?, profile_city = ?, problem = ?, today_city = ?, today_scene = ?, today_mood = ?, accessory_mode = ?, updated_at = ? where id = ? """, ( avatar_url, user_state["gender"], user_state["height"], user_state["weight"], user_state["age_info"], user_state["size"], user_state["preferred_style"], user_state["common_scenes"], user_state["profile_city"], user_state["problem"], user_state["today_city"], user_state["today_scene"], user_state["today_mood"], user_state["accessory_mode"], now, user_id, ), ) if is_male_gender(user_state["gender"]): skirt_rows = conn.execute( "select id, image_url, category, name from wardrobe_items where user_id = ?", (user_id,), ).fetchall() for row in skirt_rows: item = normalize_cloth_category({"category": row["category"], "name": row["name"]}) if is_skirt_category_name(item.get("category")): conn.execute("delete from wardrobe_items where id = ?", (row["id"],)) cleanup_upload_if_unused(conn, row["image_url"] or "") updated_user = conn.execute("select * from users where id = ?", (user_id,)).fetchone() json_response(self, {"ok": True, "profile": user_payload(updated_user)}) return if self.path == "/api/upload": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return form = cgi.FieldStorage(fp=self.rfile, headers=self.headers, environ={"REQUEST_METHOD": "POST"}) field = form["file"] if "file" in form else None if field is None or not field.filename: json_response(self, {"error": "娌℃湁鏀跺埌鍥剧墖"}, HTTPStatus.BAD_REQUEST) return try: url = save_checked_upload(field, str(user_id)) except ValueError as exc: json_response(self, {"error": str(exc)}, HTTPStatus.BAD_REQUEST) return json_response(self, {"url": url}) return if self.path == "/api/wardrobe": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) if reject_sensitive_payload(self, payload): return now = int(time.time()) with db() as conn: user_row = conn.execute("select wardrobe_limit, gender from users where id = ?", (user_id,)).fetchone() wardrobe_limit = (user_row["wardrobe_limit"] if user_row else 50) or 50 normalized_payload = normalize_cloth_category(dict(payload)) allow_duplicate = bool(payload.get("allowDuplicate")) if is_male_gender(user_row["gender"] if user_row else "") and is_skirt_category_name(normalized_payload.get("category")): json_response(self, {"error": "鐢风敓琛ｆ┍涓嶈兘褰曞叆瑁欑被鍗曞搧"}, HTTPStatus.BAD_REQUEST) return duplicate_item = None if allow_duplicate else find_duplicate_wardrobe_item(conn, user_id, normalized_payload) if duplicate_item: json_response( self, { "error": f"AI璇嗗埆鍒拌繖浠惰。鏈嶅彲鑳藉凡缁忎笂浼犺繃锛?{duplicate_item.get('name')}", "duplicateItem": duplicate_item, }, HTTPStatus.CONFLICT, ) return current_count = conn.execute("select count(*) c from wardrobe_items where user_id = ?", (user_id,)).fetchone()["c"] if current_count >= wardrobe_limit: json_response(self, {"error": f"浣犵殑琛ｆ┍鍗曞搧鏁伴噺宸茶揪涓婇檺 {wardrobe_limit} 浠讹紝璇峰厛鍒犻櫎鏃у崟鍝佹垨鑱旂郴绠＄悊鍛樿皟鏁翠笂闄"}, HTTPStatus.BAD_REQUEST) return cur = conn.execute( """ insert into wardrobe_items (user_id, name, category, color, style, fit, season, scene, difficulty, image_url, usage, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( user_id, normalized_payload.get("name", "鏈懡鍚嶅崟鍝"), normalized_payload.get("category", "涓婅。"), normalized_payload.get("color", ""), normalized_payload.get("style", ""), normalized_payload.get("fit", ""), normalized_payload.get("season", ""), normalized_payload.get("scene", ""), normalized_payload.get("difficulty", "闇€瑕佹惌閰嶆妧宸"), normalized_payload.get("image", ""), normalized_payload.get("usage", 0), now, now, ), ) row = conn.execute("select * from wardrobe_items where id = ?", (cur.lastrowid,)).fetchone() json_response(self, {"item": item_payload(row)}) return if self.path == "/api/weekly-outfit": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) plan = payload.get("plan") if not isinstance(plan, dict): json_response(self, {"error": "鏈懆绌挎惌鏁版嵁鏃犳晥"}, HTTPStatus.BAD_REQUEST) return now = int(time.time()) with db() as conn: conn.execute( """ insert into weekly_outfits (user_id, plan_json, created_at, updated_at) values (?, ?, ?, ?) on conflict(user_id) do update set plan_json=excluded.plan_json, updated_at=excluded.updated_at """, (user_id, json.dumps(plan, ensure_ascii=False), now, now), ) json_response(self, {"ok": True}) return if self.path == "/api/generation-stats": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "未登录"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) kind = payload.get("kind") try: with db() as conn: record_generation_stat(conn, user_id, kind) stats = generation_stats_payload(conn, user_id) except ValueError as exc: json_response(self, {"error": str(exc)}, HTTPStatus.BAD_REQUEST) return json_response(self, {"ok": True, "generationStats": stats}) return if self.path == "/api/activity": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "未登录"}, HTTPStatus.UNAUTHORIZED) return with db() as conn: record_user_activity(conn, user_id) json_response(self, {"ok": True}) return if self.path == "/api/ai/recognize": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) image_url = payload.get("image", "") if configured_value("DPSTAR_VISION_ENABLED", "1") == "0": result = mock_ai_result(image_url) result["note"] = "图片识别功能已关闭，请手动完善单品资料。" json_response(self, {"result": result}) return if image_url: try: result = call_openai_vision(image_url) except Exception as exc: result = mock_ai_result(image_url) result["note"] = f"AI识别接口暂不可用：{exc}。请手动完善单品资料。" json_response(self, {"result": result, "warning": result["note"]}) return json_response(self, {"result": result or mock_ai_result(image_url)}) return json_response(self, {"result": mock_ai_result(image_url)}) return if self.path == "/api/ai/outfit": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return payload = read_json(self) if reject_sensitive_payload(self, payload): return profile = payload.get("profile") or {} payload["wardrobe"] = filter_wardrobe_by_gender(payload.get("wardrobe") or [], profile.get("gender", "鐢风敓")) try: recommendation = call_ai_stylist_api(payload) except Exception as exc: json_response(self, {"error": f"AI鎼厤甯堟帴鍙ｅけ璐ワ細{exc}"}, HTTPStatus.BAD_GATEWAY) return stats = None generation_kind = payload.get("generationKind") if generation_kind in {"today", "week"}: with db() as conn: record_generation_stat(conn, user_id, generation_kind) stats = generation_stats_payload(conn, user_id) response = {"recommendation": recommendation or local_outfit_recommendation(payload)} if stats: response["generationStats"] = stats json_response(self, response) return json_response(self, {"error": "鎺ュ彛涓嶅瓨鍦"}, HTTPStatus.NOT_FOUND) def do_GET(self): request_path = urllib.parse.urlparse(self.path).path if request_path == "/api/public-settings": with db() as conn: sms_settings = conn.execute("select * from sms_settings where id = 1").fetchone() json_response(self, { "smsEnabled": bool(sms_settings["enabled"]) if sms_settings else True, "safetyEnabled": configured_value("DPSTAR_SAFETY_ENABLED", "1") != "0", "bottomMenu": read_bottom_menu(), "profileMenu": read_profile_menu(), "actionButtons": read_action_buttons(), "privacyPolicy": read_privacy_policy(), "userAgreement": read_user_agreement(), "defaultAvatarUrl": default_male_avatar_url(), "defaultMaleAvatarUrl": default_male_avatar_url(), "defaultFemaleAvatarUrl": default_female_avatar_url(), }) return if request_path == "/api/privacy-policy": json_response(self, {"privacyPolicy": read_privacy_policy()}) return if request_path == "/api/user-agreement": json_response(self, {"userAgreement": read_user_agreement()}) return if request_path == "/api/profile": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return with db() as conn: row = conn.execute("select * from users where id = ?", (user_id,)).fetchone() stats = generation_stats_payload(conn, user_id) if row else {} if not row: json_response(self, {"error": "鐢ㄦ埛涓嶅瓨鍦"}, HTTPStatus.NOT_FOUND) return profile = user_payload(row) profile["generationStats"] = stats json_response( self, { "profile": profile }, ) return if request_path == "/api/weekly-outfit": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return with db() as conn: row = conn.execute("select plan_json from weekly_outfits where user_id = ?", (user_id,)).fetchone() plan = json.loads(row["plan_json"]) if row else None json_response(self, {"plan": plan}) return if request_path == "/api/admin/overview": if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return today_start = int(time.time()) - (int(time.time()) % 86400) with db() as conn: admins = conn.execute("select id, username, permissions, created_at, updated_at from admins order by id desc").fetchall() sms_settings = conn.execute("select * from sms_settings where id = 1").fetchone() users = conn.execute("select * from users order by id desc").fetchall() items = conn.execute( """ select wardrobe_items.*, users.phone from wardrobe_items left join users on users.id = wardrobe_items.user_id order by wardrobe_items.id desc """ ).fetchall() user_count = conn.execute("select count(*) c from users").fetchone()["c"] item_count = conn.execute("select count(*) c from wardrobe_items").fetchone()["c"] today_users = conn.execute("select count(*) c from users where created_at >= ?", (today_start,)).fetchone()["c"] yesterday_users = conn.execute( "select count(*) c from users where created_at >= ? and created_at < ?", (today_start - 86400, today_start), ).fetchone()["c"] today_items = conn.execute("select count(*) c from wardrobe_items where created_at >= ?", (today_start,)).fetchone()["c"] generation_rows = conn.execute( "select kind, sum(count) c from generation_stats where stat_date = ? group by kind", (today_key(),), ).fetchall() generation_today = {row["kind"]: row["c"] or 0 for row in generation_rows} activity = admin_activity_stats(conn) upload_cleanup = cleanup_orphan_uploads(conn) upload_count = upload_cleanup["existingImages"] today_upload_count = len([p for p in UPLOAD_DIR.glob("*") if p.is_file() and p.stat().st_mtime >= today_start]) json_response( self, { "stats": { "users": user_count, "items": item_count, "uploads": upload_count, "existingImages": upload_cleanup["existingImages"], "cleanedImages": upload_cleanup["deletedImages"], "todayUsers": today_users, "yesterdayUsers": yesterday_users, "todayItems": today_items, "todayUploads": today_upload_count, "todayOutfitGenerations": generation_today.get("today", 0), "weeklyOutfitGenerations": generation_today.get("week", 0), "totalGenerationsToday": generation_today.get("today", 0) + generation_today.get("week", 0), "onlineUsers": activity["onlineUsers"], "todayActiveUsers": activity["todayActiveUsers"], "yesterdayActiveUsers": activity["yesterdayActiveUsers"], "activeUsersTwoDayTotal": activity["activeUsersTwoDayTotal"], "todayAvgStaySeconds": activity["todayAvgStaySeconds"], "yesterdayAvgStaySeconds": activity["yesterdayAvgStaySeconds"], }, "users": [user_payload(row) for row in users], "items": [ { **item_payload(row), "phone": row["phone"] or "", "createdAt": row["created_at"], "updatedAt": row["updated_at"], } for row in items ], "uploadImages": upload_gallery_payload(), "admins": [ { "id": row["id"], "username": row["username"], "permissions": row["permissions"] or "all", "createdAt": row["created_at"], "updatedAt": row["updated_at"], } for row in admins ], "smsSettings": sms_settings_payload(sms_settings), "systemSettings": system_settings_payload(), "menuSettings": read_bottom_menu(), "profileMenuSettings": read_profile_menu(), "actionButtonSettings": read_action_buttons(), "privacyPolicy": read_privacy_policy(), "userAgreement": read_user_agreement(), }, ) return if request_path == "/api/wardrobe": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return with db() as conn: rows = conn.execute("select * from wardrobe_items where user_id = ? order by id desc", (user_id,)).fetchall() json_response(self, {"items": [item_payload(row) for row in rows]}) return super().do_GET() def do_PUT(self): if self.path.startswith("/api/admin/admins/"): current_admin_id = admin_id_from_token(self) if not current_admin_id: json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return admin_id = int(self.path.rsplit("/", 1)[-1]) payload = read_json(self) username = (payload.get("username") or "").strip() password = payload.get("password") or "" permissions = (payload.get("permissions") or "all").strip() if permissions not in {"all", "users", "settings", "readonly"}: permissions = "all" if len(username) < 3: json_response(self, {"error": "绠＄悊鍛樿处鍙疯嚦灏?浣"}, HTTPStatus.BAD_REQUEST) return if password and len(password) < 6: json_response(self, {"error": "鏂板瘑鐮佽嚦灏?浣"}, HTTPStatus.BAD_REQUEST) return now = int(time.time()) try: with db() as conn: admin = conn.execute("select * from admins where id = ?", (admin_id,)).fetchone() if not admin: json_response(self, {"error": "绠＄悊鍛樹笉瀛樺湪"}, HTTPStatus.NOT_FOUND) return conn.execute( "update admins set username = ?, permissions = ?, updated_at = ? where id = ?", (username, permissions, now, admin_id), ) if password: conn.execute( "update admins set password_hash = ?, updated_at = ? where id = ?", (hash_password(password), now, admin_id), ) except sqlite3.IntegrityError: json_response(self, {"error": "绠＄悊鍛樿处鍙峰凡瀛樺湪"}, HTTPStatus.BAD_REQUEST) return json_response(self, {"ok": True}) return if self.path.startswith("/api/admin/users/"): if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return user_id = int(self.path.rsplit("/", 1)[-1]) payload = read_json(self) if reject_sensitive_payload(self, {key: value for key, value in payload.items() if key != "password"}): return phone = (payload.get("phone") or "").strip() if not phone or len(phone) != 11 or not phone.isdigit(): json_response(self, {"error": "璇疯緭鍏ユ纭殑鎵嬫満鍙"}, HTTPStatus.BAD_REQUEST) return gender = payload.get("gender") or "" height = payload.get("height") or "" weight = payload.get("weight") or "" password = payload.get("password") or "" try: wardrobe_limit = int(payload.get("wardrobeLimit") or 50) except (TypeError, ValueError): wardrobe_limit = 50 if wardrobe_limit < 1: json_response(self, {"error": "琛ｆ┍鍗曞搧涓婇檺鑷冲皯涓?1"}, HTTPStatus.BAD_REQUEST) return if password and len(password) < 6: json_response(self, {"error": "鐢ㄦ埛瀵嗙爜鑷冲皯6浣"}, HTTPStatus.BAD_REQUEST) return user_state = user_state_from_payload(payload) is_blocked = 1 if payload.get("isBlocked") else 0 now = int(time.time()) try: with db() as conn: existing_user = conn.execute("select avatar_url from users where id = ?", (user_id,)).fetchone() avatar_url = switched_default_avatar(existing_user["avatar_url"] if existing_user else "", gender) cur = conn.execute( """ update users set phone = ?, avatar_url = ?, gender = ?, height = ?, weight = ?, age_info = ?, size = ?, preferred_style = ?, common_scenes = ?, profile_city = ?, problem = ?, today_city = ?, today_scene = ?, today_mood = ?, accessory_mode = ?, wardrobe_limit = ?, is_blocked = ?, updated_at = ? where id = ? """, ( phone, avatar_url, gender, height, weight, user_state["age_info"], user_state["size"], user_state["preferred_style"], user_state["common_scenes"], user_state["profile_city"], user_state["problem"], user_state["today_city"], user_state["today_scene"], user_state["today_mood"], user_state["accessory_mode"], wardrobe_limit, is_blocked, now, user_id, ), ) if cur.rowcount == 0: json_response(self, {"error": "鐢ㄦ埛涓嶅瓨鍦"}, HTTPStatus.NOT_FOUND) return if password: conn.execute( "update users set password_hash = ?, updated_at = ? where id = ?", (hash_password(password), now, user_id), ) if is_male_gender(gender): skirt_rows = conn.execute( "select id, image_url, category, name from wardrobe_items where user_id = ?", (user_id,), ).fetchall() for row in skirt_rows: item = normalize_cloth_category({"category": row["category"], "name": row["name"]}) if is_skirt_category_name(item.get("category")): conn.execute("delete from wardrobe_items where id = ?", (row["id"],)) cleanup_upload_if_unused(conn, row["image_url"] or "") except sqlite3.IntegrityError: json_response(self, {"error": "璇ユ墜鏈哄彿宸茶鍏朵粬鐢ㄦ埛浣跨敤"}, HTTPStatus.BAD_REQUEST) return json_response(self, {"ok": True}) return if self.path.startswith("/api/admin/wardrobe/"): if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return item_id = int(self.path.rsplit("/", 1)[-1]) payload = read_json(self) if reject_sensitive_payload(self, payload): return now = int(time.time()) with db() as conn: old_row = conn.execute( """ select wardrobe_items.image_url, users.gender from wardrobe_items left join users on users.id = wardrobe_items.user_id where wardrobe_items.id = ? """, (item_id,), ).fetchone() normalized_payload = normalize_cloth_category(dict(payload)) if old_row and is_male_gender(old_row["gender"]) and is_skirt_category_name(normalized_payload.get("category")): json_response(self, {"error": "鐢风敓琛ｆ┍涓嶈兘褰曞叆瑁欑被鍗曞搧"}, HTTPStatus.BAD_REQUEST) return cur = conn.execute( """ update wardrobe_items set name=?, category=?, color=?, style=?, fit=?, season=?, scene=?, difficulty=?, image_url=?, updated_at=? where id=? """, ( normalized_payload.get("name", "鏈懡鍚嶅崟鍝"), normalized_payload.get("category", "涓婅。"), normalized_payload.get("color", ""), normalized_payload.get("style", ""), normalized_payload.get("fit", ""), normalized_payload.get("season", ""), normalized_payload.get("scene", ""), normalized_payload.get("difficulty", "闇€瑕佹惌閰嶆妧宸"), normalized_payload.get("image", ""), now, item_id, ), ) if cur.rowcount == 0: json_response(self, {"error": "鍗曞搧涓嶅瓨鍦"}, HTTPStatus.NOT_FOUND) return row = conn.execute("select * from wardrobe_items where id = ?", (item_id,)).fetchone() old_image = old_row["image_url"] if old_row else "" new_image = payload.get("image", "") if old_image and old_image != new_image: cleanup_upload_if_unused(conn, old_image) json_response(self, {"item": item_payload(row)}) return if self.path.startswith("/api/wardrobe/"): user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return item_id = int(self.path.rsplit("/", 1)[-1]) payload = read_json(self) if reject_sensitive_payload(self, payload): return now = int(time.time()) with db() as conn: old_row = conn.execute("select image_url from wardrobe_items where id = ? and user_id = ?", (item_id, user_id)).fetchone() cur = conn.execute( """ update wardrobe_items set name=?, category=?, color=?, style=?, fit=?, season=?, scene=?, difficulty=?, image_url=?, updated_at=? where id=? and user_id=? """, ( payload.get("name", "鏈懡鍚嶅崟鍝"), payload.get("category", "涓婅。"), payload.get("color", ""), payload.get("style", ""), payload.get("fit", ""), payload.get("season", ""), payload.get("scene", ""), payload.get("difficulty", "闇€瑕佹惌閰嶆妧宸"), payload.get("image", ""), now, item_id, user_id, ), ) if cur.rowcount == 0: json_response(self, {"error": "鍗曞搧涓嶅瓨鍦"}, HTTPStatus.NOT_FOUND) return row = conn.execute("select * from wardrobe_items where id = ? and user_id = ?", (item_id, user_id)).fetchone() old_image = old_row["image_url"] if old_row else "" new_image = payload.get("image", "") if old_image and old_image != new_image: cleanup_upload_if_unused(conn, old_image) json_response(self, {"item": item_payload(row)}) return json_response(self, {"error": "鎺ュ彛涓嶅瓨鍦"}, HTTPStatus.NOT_FOUND) def do_DELETE(self): if self.path.startswith("/api/admin/admins/"): current_admin_id = admin_id_from_token(self) if not current_admin_id: json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return admin_id = int(self.path.rsplit("/", 1)[-1]) if admin_id == current_admin_id: json_response(self, {"error": "涓嶈兘鍒犻櫎褰撳墠鐧诲綍鐨勭鐞嗗憳"}, HTTPStatus.BAD_REQUEST) return with db() as conn: total = conn.execute("select count(*) c from admins").fetchone()["c"] if total <= 1: json_response(self, {"error": "鑷冲皯闇€瑕佷繚鐣欎竴涓鐞嗗憳"}, HTTPStatus.BAD_REQUEST) return conn.execute("delete from admin_sessions where admin_id = ?", (admin_id,)) conn.execute("delete from admins where id = ?", (admin_id,)) json_response(self, {"ok": True}) return if self.path == "/api/account": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return with db() as conn: delete_users_and_uploads(conn, [user_id]) json_response(self, {"ok": True}) return if self.path == "/api/weekly-outfit": user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return with db() as conn: conn.execute("delete from weekly_outfits where user_id = ?", (user_id,)) json_response(self, {"ok": True}) return if self.path.startswith("/api/admin/wardrobe/"): if not is_admin(self): json_response(self, {"error": "鏃犵鐞嗗憳鏉冮檺"}, HTTPStatus.UNAUTHORIZED) return item_id = int(self.path.rsplit("/", 1)[-1]) with db() as conn: row = conn.execute("select image_url from wardrobe_items where id = ?", (item_id,)).fetchone() conn.execute("delete from wardrobe_items where id = ?", (item_id,)) if row: cleanup_upload_if_unused(conn, row["image_url"]) json_response(self, {"ok": True}) return if self.path.startswith("/api/wardrobe/"): user_id = auth_user_id(self) if not user_id: json_response(self, {"error": "鏈櫥褰"}, HTTPStatus.UNAUTHORIZED) return item_id = int(self.path.rsplit("/", 1)[-1]) with db() as conn: row = conn.execute("select image_url from wardrobe_items where id = ? and user_id = ?", (item_id, user_id)).fetchone() conn.execute("delete from wardrobe_items where id = ? and user_id = ?", (item_id, user_id)) if row: cleanup_upload_if_unused(conn, row["image_url"]) json_response(self, {"ok": True}) return json_response(self, {"error": "鎺ュ彛涓嶅瓨鍦"}, HTTPStatus.NOT_FOUND) if __name__ == "__main__": init_db() print(f"DPSTAR AI backend running: http://{HOST}:{PORT}/preview.html") ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()