from datetime import datetime, timedelta import base64 import hashlib import json import random import re import secrets import time import urllib.parse import urllib.request import uuid import io import zipfile import xml.etree.ElementTree as ET from fastapi import BackgroundTasks, Depends, FastAPI, File, Form, HTTPException, Request, Response, UploadFile, status from fastapi.middleware.cors import CORSMiddleware from sqlalchemy import func, inspect, or_, select, text from sqlalchemy.orm import Session, noload, selectinload from .config import settings from .database import Base, SessionLocal, engine, get_db from .llm import extract_questions_from_images, generate_answer_for_question, generate_questions, generate_qa from .models import AdminLoginLog, AdminUser, AnswerRecognizedQuestion, AnswerReviewDecision, AnswerScreenshot, AnswerSubmission, AnswerUser, AskAssignment, AskAssignmentBatch, AskAssignmentBatchReview, AskAssignmentBatchScreenshot, AskAssignmentQuestion, AskAssignmentReview, AskAssignmentScreenshot, AskAssignmentUsedQuestion, AskQuestion, AskTask, QALibraryEntry from .schemas import AdminUserCreate, AdminUserUpdate, LoginRequest, QAGenerateRequest, RegisterRequest from .security import create_session_token, decode_session_token, hash_password, verify_password app = FastAPI(title=settings.app_name, version="0.1.0") app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"], allow_credentials=True, allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"], allow_headers=["Content-Type"], ) # A question allocation is temporary until the user submits a completion # batch. Stale, unfinished allocations are released on the next task/status # read so the questions can flow back to the same published task link. ASK_ASSIGNMENT_TIMEOUT = timedelta(hours=5) def _release_expired_ask_assignments(db: Session, task: AskTask | None = None) -> int: """Release unsubmitted ask-task allocations older than five hours. Submitted batches are intentionally preserved (including rejected batches) because they are part of the review/history flow. Only assignments that have never submitted a batch can be safely returned to the task pool. """ cutoff = datetime.utcnow() - ASK_ASSIGNMENT_TIMEOUT statement = select(AskAssignment).where( AskAssignment.status.in_(["待选择", "进行中"]), # The five-hour window starts when the questions are claimed, not # when the user later changes a selection inside that reservation. AskAssignment.created_at < cutoff, ~AskAssignment.batches.any(), ) if task is not None: statement = statement.where(AskAssignment.task_id == task.id) expired = db.scalars(statement).unique().all() for assignment in expired: db.delete(assignment) if expired: db.flush() return len(expired) def _product_intent_error(values: list[object]) -> str | None: """Reject obvious placeholder input before creating an AI task.""" text = " ".join(str(value or "").strip() for value in values if str(value or "").strip()) if not text: return "请至少填写一项商品信息" compact = re.sub(r"[\s\W_]+", "", text, flags=re.UNICODE) if not compact: return "商品信息不能只有符号，请填写真实的商品内容" if compact.isdigit(): return "商品信息看起来像占位内容，请填写真实的商品名称、品类、功效或卖点" if re.fullmatch(r"(.{1,3})\1+", compact, flags=re.UNICODE): return "商品信息看起来是重复的占位内容，请补充真实的商品信息" if not re.search(r"[\u4e00-\u9fffA-Za-z]", compact): return "请填写包含文字的真实商品信息" return None def _generation_options_error(payload: dict, include_answers: bool = True) -> str | None: # 语言风格、目标人群、场景等生成偏好均可留空，仅限制生成条数必填。 required = [("question_count", "提问生成条数")] missing = [label for key, label in required if not str(payload.get(key, "")).strip()] if missing: return f"请填写{'、'.join(missing)}" try: question_count = int(payload.get("question_count")) except (TypeError, ValueError): return "提问生成条数必须是大于 0 的整数" if question_count < 1: return "提问生成条数必须是大于 0 的整数" return None # 回答用户链接的 MVP 数据暂存于内存：截图只在当前服务进程中保留，后续接入对象存储/持久化表。 ANSWER_TASKS = { "TASK20260822002": { "task_id": "TASK20260822002", "token": "demo-answer-token", "product_name": "泊泉雅三色修护凝精华", "status": "待上传", "questions": [ {"id": "Q001", "question": "这个精华敏感肌可以使用吗？", "answer": "质地温和，使用前建议先做局部测试。"}, {"id": "Q002", "question": "用了会不会觉得油腻？", "answer": "吸收速度较快，肤感清爽不黏腻。"}, {"id": "Q003", "question": "这款成分里面有致痘成分吗？", "answer": "配方清爽，建议结合自己的肤质进行选择。"}, {"id": "Q004", "question": "吸收速度怎么样？使用感如何？", "answer": "延展性好，吸收后肤感柔润，日常使用很舒服。"}, ], "invited_screenshots": [], "final_screenshots": [], "recognized_questions": [], "selected_question_ids": [], "user_name": "", "updated_at": datetime.utcnow().isoformat(), } } PUBLIC_ANSWER_SUBMISSIONS: list[dict] = [] PHONE_CODES: dict[str, dict] = {} DINGTALK_STATES: dict[str, float] = {} ROLE_LABELS = {"super_admin": "超级管理员", "operator": "普通运营"} ROLE_PERMISSIONS = { "super_admin": ["assistant", "library", "create", "board", "review", "users", "logs", "settings"], "operator": ["assistant", "library", "create", "board", "review"], } def public_user(user: AdminUser) -> dict: return { "id": user.id, "username": user.username, "display_name": user.display_name, "role_code": user.role_code, "role_label": ROLE_LABELS.get(user.role_code, user.role_code), "permissions": ROLE_PERMISSIONS.get(user.role_code, []), } def seed_users(db: Session): if db.scalar(select(AdminUser).limit(1)): return db.add_all([ AdminUser(username=settings.admin_username, password_hash=hash_password(settings.admin_password), display_name="系统管理员", role_code="super_admin"), AdminUser(username=settings.operator_username, password_hash=hash_password(settings.operator_password), display_name="普通运营", role_code="operator"), ]) db.commit() @app.on_event("startup") def startup(): Base.metadata.create_all(bind=engine) # Add fields introduced after the initial SQLite database was created. columns = {column["name"] for column in inspect(engine).get_columns("ask_tasks")} if "claim_questions_per_user" not in columns: with engine.begin() as connection: connection.execute(text("ALTER TABLE ask_tasks ADD COLUMN claim_questions_per_user INTEGER NOT NULL DEFAULT 5")) if "owner_id" not in columns: with engine.begin() as connection: connection.execute(text("ALTER TABLE ask_tasks ADD COLUMN owner_id INTEGER NULL")) question_columns = {column["name"] for column in inspect(engine).get_columns("answer_recognized_questions")} if "operator_image_screenshot_id" not in question_columns: with engine.begin() as connection: connection.execute(text("ALTER TABLE answer_recognized_questions ADD COLUMN operator_image_screenshot_id INTEGER NULL")) if "operator_image_screenshot_ids" not in question_columns: with engine.begin() as connection: connection.execute(text("ALTER TABLE answer_recognized_questions ADD COLUMN operator_image_screenshot_ids TEXT NULL")) db = next(get_db()) try: seed_users(db) finally: db.close() def current_user(request: Request, db: Session = Depends(get_db)) -> AdminUser: token = request.cookies.get(settings.cookie_name) payload = decode_session_token(token) if token else None if not payload: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="登录状态已失效，请重新登录") user = db.get(AdminUser, int(payload["sub"])) if not user or not user.is_active: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="账号不可用") return user def require_super_admin(user: AdminUser = Depends(current_user)) -> AdminUser: if user.role_code != "super_admin": raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="仅超级管理员可访问此功能") return user @app.get("/api/health") def health(): return {"success": True, "message": "ok"} def _answer_task(task_id: str, token: str) -> dict: task = ANSWER_TASKS.get(task_id) if not task or token != task["token"]: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="任务链接无效或已失效") return task def _public_answer_task(task: dict) -> dict: selected = set(task.get("selected_question_ids", [])) return { "task_id": task["task_id"], "product_name": task["product_name"], "status": task["status"], "recognized_questions": task.get("recognized_questions", []), "selected_questions": [q for q in task["questions"] if q["id"] in selected], "updated_at": task["updated_at"], } def _answer_user_from_token(token: str, db: Session) -> AnswerUser: user = db.scalar(select(AnswerUser).where(AnswerUser.auth_token == token)) if not user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="回答用户登录已失效") return user def _phone_number(payload: dict) -> str: phone = str(payload.get("phone", "")).strip() if not phone.isdigit() or len(phone) != 11 or not phone.startswith("1"): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请输入正确的11位手机号") return phone def _send_phone_code(payload: dict) -> dict: phone = _phone_number(payload) now = datetime.utcnow() previous = PHONE_CODES.get(phone) if previous and (now - previous["sent_at"]).total_seconds() < 60: wait = 60 - int((now - previous["sent_at"]).total_seconds()) raise HTTPException(status_code=429, detail=f"验证码发送过于频繁，请{max(1, wait)}秒后再试") code = f"{random.randint(0, 999999):06d}" PHONE_CODES[phone] = {"code": code, "sent_at": now, "expires_at": now + timedelta(minutes=5), "attempts": 0} result = {"expires_in": 300, "retry_after": 60} if settings.sms_dev_mode: result["dev_code"] = code return result def _phone_login(payload: dict, db: Session) -> AnswerUser: phone = _phone_number(payload) code = str(payload.get("code", "")).strip() challenge = PHONE_CODES.get(phone) if not challenge: raise HTTPException(status_code=400, detail="请先获取短信验证码") if challenge["attempts"] >= 5: PHONE_CODES.pop(phone, None) raise HTTPException(status_code=400, detail="验证码错误次数过多，请重新获取验证码") if datetime.utcnow() > challenge["expires_at"]: PHONE_CODES.pop(phone, None) raise HTTPException(status_code=400, detail="验证码已过期，请重新获取") if code != challenge["code"]: challenge["attempts"] += 1 raise HTTPException(status_code=400, detail="验证码错误，请重新输入") PHONE_CODES.pop(phone, None) user = db.scalar(select(AnswerUser).where(AnswerUser.phone == phone)) if not user: user = AnswerUser(public_id=f"U{uuid.uuid4().int % 10_000_000:07d}", phone=phone, auth_token=uuid.uuid4().hex) db.add(user) else: user.last_login_at = datetime.utcnow() if not user.auth_token: user.auth_token = uuid.uuid4().hex db.commit(); db.refresh(user) return user def _public_phone_session(user: AnswerUser) -> dict: return {"user_id": user.public_id, "phone": user.phone, "token": user.auth_token} def _dingtalk_configured() -> bool: return bool(settings.dingtalk_client_id and settings.dingtalk_client_secret) def _dingtalk_workbench_configured() -> bool: return bool(settings.dingtalk_corp_id and settings.dingtalk_agent_id and settings.dingtalk_client_id and settings.dingtalk_client_secret) def _dingtalk_exchange_code(code: str) -> dict: request = urllib.request.Request( "https://api.dingtalk.com/v1.0/oauth2/userAccessToken", data=json.dumps({ "clientId": settings.dingtalk_client_id, "clientSecret": settings.dingtalk_client_secret, "code": code, "grantType": "authorization_code", }).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(request, timeout=10) as response: return json.loads(response.read().decode("utf-8")) except Exception as exc: raise HTTPException(status_code=502, detail="钉钉授权码校验失败，请稍后重试") from exc def _dingtalk_user(access_token: str) -> dict: request = urllib.request.Request( "https://api.dingtalk.com/v1.0/contact/users/me", headers={"x-acs-dingtalk-access-token": access_token}, method="GET", ) try: with urllib.request.urlopen(request, timeout=10) as response: return json.loads(response.read().decode("utf-8")) except Exception as exc: raise HTTPException(status_code=502, detail="无法读取钉钉用户信息，请稍后重试") from exc def _dingtalk_workbench_user(code: str) -> dict: """Resolve a workbench JSAPI requestAuthCode code.""" token_request = urllib.request.Request( "https://oapi.dingtalk.com/gettoken?" + urllib.parse.urlencode({ "appkey": settings.dingtalk_client_id, "appsecret": settings.dingtalk_client_secret, }), method="GET", ) try: with urllib.request.urlopen(token_request, timeout=10) as response: token_data = json.loads(response.read().decode("utf-8")) except Exception as exc: raise HTTPException(status_code=502, detail="钉钉应用凭证校验失败，请稍后重试") from exc access_token = token_data.get("access_token") if not access_token: raise HTTPException(status_code=502, detail="钉钉未返回应用访问令牌，请检查应用权限") user_request = urllib.request.Request( "https://oapi.dingtalk.com/topapi/v2/user/getuserinfo?" + urllib.parse.urlencode({"access_token": access_token}), data=json.dumps({"code": code}).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(user_request, timeout=10) as response: user_data = json.loads(response.read().decode("utf-8")) except Exception as exc: raise HTTPException(status_code=502, detail="无法读取钉钉用户信息，请检查应用免登权限") from exc if user_data.get("errcode", 0) not in (0, None): raise HTTPException(status_code=502, detail=f"钉钉用户信息读取失败：{user_data.get('errmsg', '请检查应用权限')}") return user_data.get("result") or user_data def _dingtalk_create_session(profile: dict, db: Session, response: Response) -> AdminUser: union_id = str(profile.get("unionId") or profile.get("unionid") or profile.get("openId") or profile.get("userid") or profile.get("userId") or "").strip() if not union_id: raise HTTPException(status_code=502, detail="钉钉用户信息缺少唯一标识") if settings.dingtalk_allowed_union_ids and union_id not in settings.dingtalk_allowed_union_ids: raise HTTPException(status_code=403, detail="当前钉钉账号未被授权使用运营后台") username = f"dingtalk_{union_id}"[:64] user = db.scalar(select(AdminUser).where(AdminUser.username == username)) display_name = str(profile.get("nick") or profile.get("name") or "钉钉用户").strip()[:64] role_code = "super_admin" if union_id in settings.dingtalk_super_admin_union_ids else "operator" if not user: user = AdminUser(username=username, password_hash=hash_password(secrets.token_urlsafe(32)), display_name=display_name, role_code=role_code) db.add(user) elif not user.is_active: raise HTTPException(status_code=403, detail="运营账号已被停用") else: user.display_name = display_name or user.display_name if union_id in settings.dingtalk_super_admin_union_ids: user.role_code = "super_admin" user.last_login_at = datetime.utcnow() db.flush() db.add(AdminLoginLog(user_id=user.id, username=username, success=True)) db.commit(); db.refresh(user) response.set_cookie(settings.cookie_name, create_session_token(user.id, user.role_code), httponly=True, secure=settings.cookie_secure, samesite="lax", max_age=settings.session_hours * 3600) return user def _dingtalk_login(code: str, db: Session, response: Response) -> AdminUser: if not _dingtalk_configured(): raise HTTPException(status_code=503, detail="钉钉登录尚未配置，请联系管理员") code = code.strip() if not code: raise HTTPException(status_code=400, detail="钉钉授权码不能为空") token_data = _dingtalk_exchange_code(code) access_token = token_data.get("accessToken") or token_data.get("access_token") if not access_token: raise HTTPException(status_code=502, detail="钉钉未返回有效授权令牌") profile = _dingtalk_user(access_token) return _dingtalk_create_session(profile, db, response) def _can_access_ask_task(task: AskTask, user: AdminUser) -> bool: return user.role_code == "super_admin" or task.owner_id == user.id def _require_ask_task_access(task: AskTask | None, user: AdminUser) -> AskTask: if not task or not _can_access_ask_task(task, user): raise HTTPException(status_code=404, detail="提问任务不存在") return task def _ask_task_owner_clause(user: AdminUser): return True if user.role_code == "super_admin" else AskTask.owner_id == user.id def _published_ask_task(task_no: str, db: Session, *, lock: bool = False) -> AskTask: if task_no == "latest": statement = select(AskTask).where(AskTask.status == "已发布").order_by(AskTask.created_at.desc()) else: statement = select(AskTask).where(AskTask.task_no == task_no, AskTask.status == "已发布") # Serialize allocation for the same task. This prevents two users who # open the link at the same moment from receiving the same question. if lock: statement = statement.with_for_update() task = db.scalar(statement) if not task: raise HTTPException(status_code=404, detail="提问任务链接无效、未发布或已结束") return task def _ask_assignment_data(assignment: AskAssignment, include_images: bool = True) -> dict: items = sorted(assignment.items, key=lambda item: item.question.sequence_no) used_ids = {item.question_id for item in assignment.used_questions} # List endpoints must not touch the screenshot relationship. The image # BLOBs are large and loading them for every row was the main cause of the # multi-second task-board/review-center delay. screenshots = list(assignment.screenshots) if include_images else [] question_screenshots = [item for item in screenshots if item.stage == "question"] payment_screenshots = [item for item in screenshots if item.stage == "payment"] latest_review = ( max(assignment.review_decisions, key=lambda item: item.created_at) if include_images and assignment.review_decisions else None ) # Backfill the read model for approvals created before the used-question # table was introduced. New approvals are persisted in that table below. if assignment.status in {"待审核", "已通过"}: # Submission itself consumes the selected questions. The fallback # also makes records created before the used-question table behave # correctly without a data migration. used_ids.update(item.question_id for item in items if item.selected) return { "assignment_id": assignment.id, "status": assignment.status, "max_questions": len(items), "used_count": len(used_ids), "remaining_count": max(len(items) - len(used_ids), 0), "task_id": assignment.task.task_no, "product_name": assignment.task.product_name, "product_link": assignment.task.product_link, "user_id": assignment.user.public_id, "phone": assignment.user.phone, "created_at": assignment.created_at.isoformat(), "updated_at": assignment.updated_at.isoformat(), "question_screenshots": [{"id": image.id, "name": image.filename, "data_url": f"data:{image.content_type};base64,{base64.b64encode(image.image_data).decode()}"} for image in question_screenshots] if include_images else [], "payment_screenshots": [{"id": image.id, "name": image.filename, "data_url": f"data:{image.content_type};base64,{base64.b64encode(image.image_data).decode()}"} for image in payment_screenshots] if include_images else [], "latest_review": ({ "approved": latest_review.approved, "reason_code": latest_review.reason_code, "reason_text": latest_review.reason_text, "created_at": latest_review.created_at.isoformat(), } if latest_review else None), "questions": [{ "id": item.question.id, "sequence_no": item.question.sequence_no, "question": item.question.question_text, "selected": item.selected, "used": item.question_id in used_ids, } for item in items], } def _ask_batch_data(batch: AskAssignmentBatch, include_images: bool = True) -> dict: """Serialize one independent submission batch for review/history views.""" assignment = batch.assignment parent = _ask_assignment_data(assignment, include_images=False) try: selected_ids = {int(value) for value in (batch.selected_question_ids or "").split(",") if value} except ValueError: selected_ids = set() used_ids = {item.question_id for item in assignment.used_questions} latest_review = ( max(batch.review_decisions, key=lambda item: item.created_at) if include_images and batch.review_decisions else None ) screenshots = list(batch.screenshots) if include_images else [] question_images = [item for item in screenshots if item.stage == "question"] payment_images = [item for item in screenshots if item.stage == "payment"] questions = [] for item in sorted(assignment.items, key=lambda value: value.question.sequence_no): questions.append({ "id": item.question.id, "sequence_no": item.question.sequence_no, "question": item.question.question_text, "selected": item.question_id in selected_ids, "used": item.question_id in used_ids, }) data = { **parent, "assignment_id": batch.id, "batch_id": batch.id, "parent_assignment_id": assignment.id, # A paid batch remains an approved audit record in the review center; # payout completion is represented separately on the task board. "status": {"pending": "待审核", "approved": "已通过", "paid": "已通过", "rejected": "未通过"}.get(batch.status, batch.status), "created_at": batch.created_at.isoformat(), "updated_at": batch.updated_at.isoformat(), "questions": questions, "question_screenshots": [ {"id": image.id, "name": image.filename, "data_url": f"data:{image.content_type};base64,{base64.b64encode(image.image_data).decode()}"} for image in question_images ] if include_images else [], "payment_screenshots": [ {"id": image.id, "name": image.filename, "data_url": f"data:{image.content_type};base64,{base64.b64encode(image.image_data).decode()}"} for image in payment_images ] if include_images else [], "latest_review": ({ "approved": latest_review.approved, "reason_code": latest_review.reason_code, "reason_text": latest_review.reason_text, "created_at": latest_review.created_at.isoformat(), } if latest_review else None), } return data def _sync_ask_assignment_status(assignment: AskAssignment) -> None: """Keep the aggregate assignment status aligned with its latest batch.""" if not assignment.batches: return # Batch order means submission order; a later review decision must not # make an older batch replace the user's newest pending batch. latest = max(assignment.batches, key=lambda value: value.created_at) assignment.status = {"pending": "待审核", "approved": "已通过", "rejected": "未通过"}.get(latest.status, latest.status) assignment.updated_at = datetime.utcnow() def _image_url(image: AnswerScreenshot) -> str: return f"data:{image.content_type};base64,{base64.b64encode(image.image_data).decode()}" def _submission_data(record: AnswerSubmission, include_images: bool = True) -> dict: # List endpoints deliberately omit screenshot BLOBs. Do not even touch # the lazy relationship in that mode, otherwise SQLAlchemy fetches every # image from remote MySQL before returning the lightweight response. screenshots = list(record.screenshots) if include_images else [] invited = [x for x in screenshots if x.stage == "invited"] completion = [x for x in screenshots if x.stage == "completion"] payment = [x for x in screenshots if x.stage == "payment"] images = {x.id: x for x in screenshots} latest_review = max(record.review_decisions, key=lambda value: value.created_at) if record.review_decisions else None return { "id": record.id, "submission_no": record.submission_no, "user_id": record.user.public_id, "phone": record.user.phone, "status": record.status, "created_at": record.created_at.isoformat(), "updated_at": record.updated_at.isoformat(), "invited_screenshots": [{"id": x.id, "name": x.filename, "data_url": _image_url(x)} for x in invited] if include_images else [], "completion_screenshots": [{"id": x.id, "name": x.filename, "data_url": _image_url(x)} for x in completion] if include_images else [], "payment_screenshots": [{"id": x.id, "name": x.filename, "data_url": _image_url(x)} for x in payment] if include_images else [], "latest_review": ({ "approved": latest_review.approved, "reason_code": latest_review.reason_code, "reason_text": latest_review.reason_text, "created_at": latest_review.created_at.isoformat(), } if latest_review else None), "questions": [{ "id": q.id, "sequence_no": q.sequence_no, "product_title": q.product_title, "product_image": _image_url(images[q.product_image_screenshot_id]) if include_images and q.product_image_screenshot_id in images else None, "source_image": _image_url(images[q.source_screenshot_id]) if include_images and q.source_screenshot_id in images else None, "operator_images": [ _image_url(images[int(image_id)]) for image_id in ((q.operator_image_screenshot_ids or str(q.operator_image_screenshot_id or "")).split(",")) if image_id.strip().isdigit() and int(image_id) in images ] if include_images else [], "question": q.question_text, "answer": q.answer_text, "answer_status": q.answer_status, "selected": q.selected, } for q in sorted(record.questions, key=lambda value: value.sequence_no)], } def _task_board_data(record: AnswerSubmission, include_images: bool = True) -> dict: selected = [q for q in sorted(record.questions, key=lambda value: value.sequence_no) if q.selected] product_titles = list(dict.fromkeys(q.product_title for q in selected if q.product_title)) summary = _submission_data(record, include_images=True) if include_images else { "id": record.id, "submission_no": record.submission_no, "user_id": record.user.public_id, "phone": record.user.phone, "status": record.status, "created_at": record.created_at.isoformat(), "updated_at": record.updated_at.isoformat(), "invited_screenshots": [], "completion_screenshots": [], "payment_screenshots": [], "latest_review": None, "questions": [], } status_labels = { "待用户完成": "进行中", "待审核": "待审核", # Passing the content audit means the submission is ready for the # finance check; it is not fully complete until the refund is handled. "已通过": "待返款", "已完成": "已完成", "未通过": "进行中", "已撤销": "已撤销", } completed_count = len(selected) if record.status in {"待审核", "已通过", "已完成"} else 0 return { **summary, "board_key": f"answer-{record.id}", "record_kind": "answer_submission", "task_id": record.submission_no, "task_type": "answer", "task_type_label": "回答任务", "status_label": status_labels.get(record.status, record.status), "product_name": " / ".join(product_titles) or "待识别商品", "qa_count": len(selected), "claimed_count": 1, "submitted_count": 1 if record.status in {"待审核", "已通过", "已完成"} else 0, "pending_review_count": 1 if record.status == "待审核" else 0, "completed_count": completed_count, "total_count": len(selected), "creator_name": "运营后台", "can_cancel": record.status in {"待用户完成", "待审核", "未通过"}, "can_delete": True, "delete_endpoint": f"/api/admin/task-board/{record.id}", "refund_endpoint": f"/api/admin/task-board/{record.id}/confirm-refund" if record.status == "已通过" else None, "assigned_users": [{"user_id": record.user.public_id, "phone": record.user.phone, "status": status_labels.get(record.status, record.status)}], "selected_questions": ([{"id": q.id, "sequence_no": q.sequence_no, "question": q.question_text, "answer": q.answer_text, "product_title": q.product_title} for q in selected] if include_images else []), } def _ask_assignment_child_data(assignment: AskAssignment) -> dict: """Return a compact user-claim row nested under its system ask task.""" task = assignment.task selected_ids = { item.question_id for item in assignment.items if item.selected } used_ids = {item.question_id for item in assignment.used_questions} latest_batch = max(assignment.batches, key=lambda value: value.updated_at or value.created_at, default=None) batch_selected_ids = set() if latest_batch and latest_batch.selected_question_ids: batch_selected_ids = { int(value) for value in latest_batch.selected_question_ids.split(",") if value.strip().isdigit() } question_count = len(batch_selected_ids or selected_ids or used_ids) refund_completed = bool(latest_batch and latest_batch.status == "paid") or assignment.status == "已完成" refund_pending = bool(latest_batch and latest_batch.status == "approved") or assignment.status == "已通过" if refund_completed: status_label = "已完成" elif refund_pending: status_label = "待返款" elif assignment.status == "待审核" or (latest_batch and latest_batch.status == "pending"): status_label = "待审核" elif assignment.status == "已撤销": status_label = "已撤销" else: status_label = "进行中" detail_endpoint = None refund_endpoint = None if latest_batch and latest_batch.status in {"approved", "paid"}: detail_endpoint = f"/api/admin/task-board/ask-batches/{latest_batch.id}" if not refund_completed: refund_endpoint = f"/api/admin/task-board/ask-batches/{latest_batch.id}/confirm-refund" elif assignment.status in {"已通过", "已完成"} and not assignment.batches: detail_endpoint = f"/api/admin/task-board/ask-assignments/{assignment.id}" if not refund_completed: refund_endpoint = f"/api/admin/task-board/ask-assignments/{assignment.id}/confirm-refund" return { "id": assignment.id, "board_key": f"ask-assignment-{assignment.id}", "record_kind": "ask_assignment", "is_child": True, "task_id": task.task_no, "task_type": "ask", "task_type_label": "提问任务", "product_name": task.product_name, "product_link": task.product_link, "status": assignment.status, "status_label": status_label, "user_id": assignment.user.public_id, "phone": assignment.user.phone, "qa_count": question_count, "claimed_count": question_count, "submitted_count": len(used_ids), "pending_review_count": 1 if status_label == "待审核" else 0, "completed_count": len(used_ids) if used_ids else (question_count if refund_completed else 0), "total_count": question_count, "creator_name": task.owner.display_name if task.owner else "运营后台", "created_at": assignment.created_at.isoformat(), "updated_at": assignment.updated_at.isoformat(), "can_cancel": False, "can_complete": False, "can_delete": False, "detail_endpoint": detail_endpoint, "refund_endpoint": refund_endpoint, "assigned_users": [{"user_id": assignment.user.public_id, "phone": assignment.user.phone, "status": status_label}], "selected_questions": [], "question_screenshots": [], "completion_screenshots": [], "payment_screenshots": [], "latest_review": None, } def _ask_task_data(task: AskTask, include_questions: bool = True) -> dict: questions = sorted(task.questions, key=lambda value: value.sequence_no) selected_questions = [q for q in questions if q.selected] claimed_tasks = sorted( (_ask_assignment_child_data(assignment) for assignment in task.assignments), key=lambda value: value.get("updated_at") or value.get("created_at") or "", reverse=True, ) assigned_count = len(task.assignments) completed_count = sum(len(assignment.used_questions) for assignment in task.assignments) pending_count = sum(child["pending_review_count"] for child in claimed_tasks) return { "id": task.id, "board_key": f"ask-task-{task.id}", "record_kind": "ask_task", "task_id": task.task_no, "task_type": "ask", "task_type_label": "提问任务", "product_name": task.product_name, "product_link": task.product_link, "status": task.status, "claim_questions_per_user": task.claim_questions_per_user or 5, "status_label": {"已发布": "进行中", "已撤销": "已撤销", "已完成": "已完成"}.get(task.status, task.status), "qa_count": len(selected_questions), "claimed_count": assigned_count, "submitted_count": completed_count, "pending_review_count": pending_count, "completed_count": completed_count, "total_count": len(selected_questions), "creator_name": task.owner.display_name if task.owner else "运营后台", "created_at": task.created_at.isoformat(), "updated_at": task.updated_at.isoformat(), "can_cancel": task.status == "已发布", "can_complete": task.status == "已发布", "can_delete": True, "delete_endpoint": f"/api/admin/task-board/ask-tasks/{task.id}", "user_id": "--", "phone": "", "questions": [], "selected_questions": ([{"id": q.id, "sequence_no": q.sequence_no, "question": q.question_text, "answer": "", "product_title": task.product_name} for q in selected_questions] if include_questions else []), "assigned_users": [child["assigned_users"][0] for child in claimed_tasks if child.get("assigned_users")], "claimed_tasks": claimed_tasks, "invited_screenshots": [], "completion_screenshots": [], "payment_screenshots": [], "latest_review": None, "detail_endpoint": f"/api/admin/task-board/ask-tasks/{task.id}" if not include_questions else None, } def _answer_review_data(record: AnswerSubmission, include_images: bool = True) -> dict: """Keep completed refunds visible as approved audit history.""" data = _submission_data(record, include_images=include_images) if record.status == "已完成": data["status"] = "已通过" data["refund_completed"] = True else: data["refund_completed"] = False return data def _ask_review_board_data(record: AskAssignmentBatch | AskAssignment, include_images: bool = False) -> dict: """Convert one approved ask submission into an independent payout row.""" is_batch = isinstance(record, AskAssignmentBatch) assignment = record.assignment if is_batch else record if include_images: data = _ask_batch_data(record, include_images=True) if is_batch else _ask_assignment_data(record, include_images=True) selected = [question for question in data["questions"] if question["selected"]] elif is_batch: selected_ids = {int(value) for value in (record.selected_question_ids or "").split(",") if value} selected = [] data = { "assignment_id": record.id, "status": "已通过", "task_id": assignment.task.task_no, "product_name": assignment.task.product_name, "product_link": assignment.task.product_link, "user_id": assignment.user.public_id, "phone": assignment.user.phone, "created_at": record.created_at.isoformat(), "updated_at": record.updated_at.isoformat(), "questions": [], "question_screenshots": [], "payment_screenshots": [], "latest_review": None, } else: data = _ask_assignment_data(record, include_images=False) selected = [question for question in data["questions"] if question["selected"]] selected_ids = {question["id"] for question in selected} data["questions"] = [] refund_completed = record.status == "paid" if is_batch else record.status == "已完成" selected_count = len(selected) if include_images else len(selected_ids) return { **data, "id": record.id, "board_key": f"ask-{'batch' if is_batch else 'assignment'}-{record.id}", "record_kind": "ask_batch" if is_batch else "ask_assignment", "task_type": "ask", "task_type_label": "提问任务", "status_label": "已完成" if refund_completed else "待返款", "product_name": assignment.task.product_name, "qa_count": selected_count, "claimed_count": selected_count, "submitted_count": selected_count, "pending_review_count": 0, "completed_count": selected_count, "total_count": selected_count, "creator_name": "运营后台", "can_cancel": False, "can_delete": True, "delete_endpoint": ( f"/api/admin/task-board/ask-batches/{record.id}" if is_batch else f"/api/admin/task-board/ask-assignments/{record.id}" ), "user_id": assignment.user.public_id, "phone": assignment.user.phone, "assigned_users": [{ "user_id": assignment.user.public_id, "phone": assignment.user.phone, "status": "已完成" if refund_completed else "待返款", }], "selected_questions": ([{ "id": question["id"], "sequence_no": question["sequence_no"], "question": question["question"], "answer": "", "product_title": assignment.task.product_name, } for question in selected] if include_images else []), "detail_endpoint": ( f"/api/admin/task-board/ask-batches/{record.id}" if is_batch else f"/api/admin/task-board/ask-assignments/{record.id}" ), "refund_endpoint": ( None if refund_completed else ( f"/api/admin/task-board/ask-batches/{record.id}/confirm-refund" if is_batch else f"/api/admin/task-board/ask-assignments/{record.id}/confirm-refund" ) ), } @app.post("/api/answer/auth/phone-login") def answer_phone_login(payload: dict, db: Session = Depends(get_db)): return {"success": True, "data": _public_phone_session(_phone_login(payload, db))} @app.post("/api/answer/auth/send-code") def answer_send_code(payload: dict): return {"success": True, "data": _send_phone_code(payload)} @app.post("/api/ask/auth/phone-login") def ask_phone_login(payload: dict, db: Session = Depends(get_db)): return {"success": True, "data": _public_phone_session(_phone_login(payload, db))} @app.post("/api/ask/auth/send-code") def ask_send_code(payload: dict): return {"success": True, "data": _send_phone_code(payload)} @app.get("/api/ask/tasks/{task_no}") def public_ask_task(task_no: str, token: str, db: Session = Depends(get_db)): user = _answer_user_from_token(token, db) task = _published_ask_task(task_no, db, lock=True) released = _release_expired_ask_assignments(db, task) if released: db.commit() db.expire_all() task = _published_ask_task(task_no, db, lock=True) assignment = db.scalar(select(AskAssignment).where(AskAssignment.task_id == task.id, AskAssignment.user_id == user.id)) if not assignment: # Keep each question reserved while the user is working. Expired # unfinished reservations have already been removed above, so these # are the currently allocated questions that must stay excluded. allocated_ids = set(db.scalars( select(AskAssignmentQuestion.question_id) .join(AskAssignment, AskAssignmentQuestion.assignment_id == AskAssignment.id) .where(AskAssignment.task_id == task.id) ).all()) available = [ question for question in task.questions if question.selected and question.id not in allocated_ids ] if not available: raise HTTPException( status_code=409, detail="本次任务已被领取完，请关注下一个任务！", ) claim_limit = task.claim_questions_per_user or 5 assigned = random.sample(available, min(claim_limit, len(available))) assignment = AskAssignment(task_id=task.id, user_id=user.id, status="待选择") db.add(assignment); db.flush() for question in assigned: db.add(AskAssignmentQuestion(assignment_id=assignment.id, question_id=question.id, selected=False)) db.commit(); db.refresh(assignment) latest_batch = db.scalar( select(AskAssignmentBatch) .where(AskAssignmentBatch.assignment_id == assignment.id) .order_by(AskAssignmentBatch.created_at.desc()) ) # While the user is selecting a new batch, the parent assignment contains # the current selection. Do not replace it with the previous submitted # batch or a refresh would appear to undo the user's choices. if latest_batch and assignment.status != "进行中": return {"success": True, "data": _ask_batch_data(latest_batch)} return {"success": True, "data": _ask_assignment_data(assignment)} @app.get("/api/ask/me/assignments") def public_ask_assignments(token: str, db: Session = Depends(get_db)): user = _answer_user_from_token(token, db) if _release_expired_ask_assignments(db): db.commit() records = db.scalars( select(AskAssignment) .where(AskAssignment.user_id == user.id) .order_by(AskAssignment.updated_at.desc()) ).unique().all() data = [] for record in records: batches = sorted(record.batches, key=lambda value: value.created_at, reverse=True) if batches: data.extend(_ask_batch_data(batch, include_images=False) for batch in batches) else: data.append(_ask_assignment_data(record, include_images=False)) return {"success": True, "data": data} @app.post("/api/ask/tasks/{task_no}/selection") def select_public_ask_questions(task_no: str, payload: dict, db: Session = Depends(get_db)): user = _answer_user_from_token(str(payload.get("token", "")), db) task = _published_ask_task(task_no, db) if _release_expired_ask_assignments(db, task): db.commit() assignment = db.scalar(select(AskAssignment).where(AskAssignment.task_id == task.id, AskAssignment.user_id == user.id)) if not assignment: raise HTTPException(status_code=409, detail="请先领取提问任务") try: selected_ids = {int(value) for value in payload.get("question_ids", [])} except (TypeError, ValueError): raise HTTPException(status_code=400, detail="问题选择数据无效") assigned_ids = {item.question_id for item in assignment.items} if not selected_ids: raise HTTPException(status_code=400, detail="请至少选择一个问题") if not selected_ids.issubset(assigned_ids): raise HTTPException(status_code=400, detail="选择中包含未分配给当前用户的问题") used_ids = {item.question_id for item in assignment.used_questions} if selected_ids & used_ids: raise HTTPException(status_code=400, detail="已认领的问题不能重复选择") for item in assignment.items: # The selection represents only the next batch. Questions in # ``used_ids`` remain locked/read-only and are not resubmitted. item.selected = item.question_id in selected_ids assignment.status = "进行中" db.commit(); db.refresh(assignment) return {"success": True, "data": _ask_assignment_data(assignment)} @app.post("/api/ask/tasks/{task_no}/submission") async def submit_public_ask_task( task_no: str, token: str = Form(...), files: list[UploadFile] = File(...), payment_file: UploadFile = File(...), payee_name: str = Form(default=""), db: Session = Depends(get_db), ): user = _answer_user_from_token(token, db) task = _published_ask_task(task_no, db) if _release_expired_ask_assignments(db, task): db.commit() assignment = db.scalar(select(AskAssignment).where(AskAssignment.task_id == task.id, AskAssignment.user_id == user.id)) if not assignment: raise HTTPException(status_code=404, detail="提问任务不存在") if assignment.status not in {"进行中", "未通过"}: raise HTTPException(status_code=409, detail="当前任务不能提交审核") if not any(item.selected for item in assignment.items): raise HTTPException(status_code=400, detail="请至少选择一个问题") if not files: raise HTTPException(status_code=400, detail="请至少上传一张提问截图") allowed = {"image/jpeg", "image/png", "image/webp"} if any(upload.content_type not in allowed for upload in [*files, payment_file]): raise HTTPException(status_code=400, detail="仅支持 JPG、PNG、WEBP 图片") selected_ids = [item.question_id for item in assignment.items if item.selected] batch = AskAssignmentBatch( assignment_id=assignment.id, status="pending", selected_question_ids=",".join(str(value) for value in selected_ids), ) db.add(batch) db.flush() for upload in files: db.add(AskAssignmentBatchScreenshot(batch_id=batch.id, stage="question", filename=upload.filename or "question", content_type=upload.content_type, image_data=await upload.read())) db.add(AskAssignmentBatchScreenshot(batch_id=batch.id, stage="payment", filename=payment_file.filename or "payment", content_type=payment_file.content_type, image_data=await payment_file.read())) assignment.status = "待审核" used_ids = {item.question_id for item in assignment.used_questions} for item in assignment.items: if item.selected and item.question_id not in used_ids: db.add(AskAssignmentUsedQuestion(assignment_id=assignment.id, question_id=item.question_id)) used_ids.add(item.question_id) assignment.updated_at = datetime.utcnow() db.commit(); db.refresh(batch); db.refresh(assignment) return {"success": True, "data": _ask_batch_data(batch)} @app.get("/api/admin/ask-completion-reviews") def ask_completion_reviews(user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): batches = db.scalars( select(AskAssignmentBatch).join(AskAssignmentBatch.assignment).join(AskAssignment.task) .where(AskAssignmentBatch.status == "pending") .where(_ask_task_owner_clause(user)) .order_by(AskAssignmentBatch.updated_at.desc()) ).unique().all() legacy = db.scalars( select(AskAssignment).join(AskAssignment.task).where( AskAssignment.status == "待审核", ~AskAssignment.batches.any(), AskAssignment.screenshots.any(), _ask_task_owner_clause(user), ).order_by(AskAssignment.updated_at.desc()) ).unique().all() data = [_ask_batch_data(batch, include_images=False) for batch in batches] data.extend(_ask_assignment_data(record, include_images=False) for record in legacy) return {"success": True, "data": data} @app.get("/api/admin/review-pending-count") def review_pending_count(user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): """Return lightweight sidebar counts without loading review rows or images.""" answer_count = db.scalar( select(func.count(AnswerSubmission.id)).where(AnswerSubmission.status == "待审核") ) or 0 ask_batch_count = db.scalar( select(func.count(AskAssignmentBatch.id)).join(AskAssignmentBatch.assignment).join(AskAssignment.task).where(AskAssignmentBatch.status == "pending", _ask_task_owner_clause(user)) ) or 0 legacy_ask_count = db.scalar( select(func.count(AskAssignment.id)).join(AskAssignment.task).where( AskAssignment.status == "待审核", ~AskAssignment.batches.any(), AskAssignment.screenshots.any(), _ask_task_owner_clause(user), ) ) or 0 ask_count = ask_batch_count + legacy_ask_count return { "success": True, "data": { "ask": ask_count, "answer": answer_count, "total": ask_count + answer_count, }, } @app.get("/api/admin/ask-completion-reviews/{assignment_id}") def ask_completion_review_detail(assignment_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): batch = db.get(AskAssignmentBatch, assignment_id) if batch: _require_ask_task_access(batch.assignment.task, user) return {"success": True, "data": _ask_batch_data(batch)} assignment = db.get(AskAssignment, assignment_id) if not assignment or not _can_access_ask_task(assignment.task, user): raise HTTPException(status_code=404, detail="提问审核任务不存在") return {"success": True, "data": _ask_assignment_data(assignment)} @app.post("/api/admin/ask-completion-reviews/{assignment_id}/decision") def ask_completion_decision(assignment_id: int, payload: dict, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): batch = db.get(AskAssignmentBatch, assignment_id) if batch: _require_ask_task_access(batch.assignment.task, user) if batch.status != "pending": raise HTTPException(status_code=409, detail="review already completed") approved = bool(payload.get("approved")) reason_code = str(payload.get("reason_code", "")).strip() or None reason_text = str(payload.get("reason_text", "")).strip() or None if not approved and not (reason_code or reason_text): raise HTTPException(status_code=400, detail="review reason is required") db.add(AskAssignmentBatchReview(batch_id=batch.id, approved=approved, reason_code=reason_code, reason_text=reason_text)) batch.status = "approved" if approved else "rejected" _sync_ask_assignment_status(batch.assignment) db.commit(); db.refresh(batch) return {"success": True, "data": _ask_batch_data(batch)} assignment = db.get(AskAssignment, assignment_id) if not assignment: raise HTTPException(status_code=404, detail="提问审核任务不存在") if assignment.status != "待审核": raise HTTPException(status_code=409, detail="任务不在待审核状态") approved = bool(payload.get("approved")) reason_code = str(payload.get("reason_code", "")).strip() or None reason_text = str(payload.get("reason_text", "")).strip() or None if not approved and not (reason_code or reason_text): raise HTTPException(status_code=400, detail="请选择或填写审核不通过原因") db.add(AskAssignmentReview(assignment_id=assignment.id, approved=approved, reason_code=reason_code, reason_text=reason_text)) if approved: used_ids = {item.question_id for item in assignment.used_questions} for item in assignment.items: if item.selected and item.question_id not in used_ids: db.add(AskAssignmentUsedQuestion(assignment_id=assignment.id, question_id=item.question_id)) used_ids.add(item.question_id) assignment.status = "已通过" if approved else "未通过" assignment.updated_at = datetime.utcnow() db.commit(); db.refresh(assignment) return {"success": True, "data": _ask_assignment_data(assignment)} @app.get("/api/answer/me/submissions") def answer_my_submissions(token: str, include_images: bool = False, db: Session = Depends(get_db)): user = _answer_user_from_token(token, db) records = db.scalars(select(AnswerSubmission).where(AnswerSubmission.user_id == user.id).order_by(AnswerSubmission.created_at.desc())).unique().all() return {"success": True, "data": [_submission_data(record, include_images=include_images) for record in records]} @app.get("/api/answer/me/submissions/{submission_id}") def answer_submission_detail(submission_id: int, token: str, db: Session = Depends(get_db)): user = _answer_user_from_token(token, db) record = db.get(AnswerSubmission, submission_id) if not record or record.user_id != user.id: raise HTTPException(status_code=404, detail="任务不存在") return {"success": True, "data": _submission_data(record, include_images=True)} def _process_answer_submission(submission_id: int): """Run OCR and answer drafting after the upload response has been sent.""" db = SessionLocal() try: record = db.get(AnswerSubmission, submission_id) if not record or record.status != "待筛选": return saved = [shot for shot in record.screenshots if shot.stage == "invited"] extracted = extract_questions_from_images([(shot.image_data, shot.content_type) for shot in saved]) if not extracted: extracted = [ {"product_title": "待识别商品", "question": "这款商品使用起来感觉怎么样？"}, {"product_title": "待识别商品", "question": "敏感肌可以使用这款商品吗？"}, {"product_title": "待识别商品", "question": "使用后的肤感和效果怎么样？"}, ][:max(1, min(3, len(saved)))] for index, item in enumerate(extracted): product, question = item["product_title"], item["question"] db.add(AnswerRecognizedQuestion( submission_id=record.id, source_screenshot_id=saved[index % len(saved)].id, product_image_screenshot_id=saved[index % len(saved)].id, sequence_no=index + 1, product_title=product, question_text=question, answer_text=generate_answer_for_question(product, question), answer_status="已生成", )) record.updated_at = datetime.utcnow() db.commit() except Exception: db.rollback() # Keep the submission visible to operators even if OCR/model is unavailable. try: record = db.get(AnswerSubmission, submission_id) saved = [shot for shot in (record.screenshots if record else []) if shot.stage == "invited"] if record and not record.questions and saved: for index, question in enumerate(( "这款商品使用起来感觉怎么样？", "敏感肌可以使用这款商品吗？", "使用后的肤感和效果怎么样？", )[:max(1, min(3, len(saved)))]): db.add(AnswerRecognizedQuestion( submission_id=record.id, source_screenshot_id=saved[index % len(saved)].id, product_image_screenshot_id=saved[index % len(saved)].id, sequence_no=index + 1, product_title="待识别商品", question_text=question, answer_text="建议结合自身需求和肤质先少量试用。", answer_status="待补充", )) db.commit() except Exception: db.rollback() finally: db.close() @app.post("/api/answer/me/submissions") async def answer_create_submission( background_tasks: BackgroundTasks, token: str = Form(...), files: list[UploadFile] = File(...), db: Session = Depends(get_db), ): user = _answer_user_from_token(token, db) if not files or len(files) > 10: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="最多上传10张截图") # 先落库并立即返回；OCR/回答生成放到响应后的后台任务执行。 record = AnswerSubmission(submission_no=f"ANS{datetime.utcnow().strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:4].upper()}", user_id=user.id, status="待筛选") db.add(record); db.flush() saved = [] for upload in files: if upload.content_type not in {"image/jpeg", "image/png", "image/webp"}: db.rollback(); raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 JPG、PNG、WEBP 图片") shot = AnswerScreenshot(submission_id=record.id, stage="invited", filename=upload.filename or "screenshot", content_type=upload.content_type, image_data=await upload.read()) db.add(shot); db.flush(); saved.append(shot) db.commit(); db.refresh(record) background_tasks.add_task(_process_answer_submission, record.id) # 先返回轻量结果，图片由后续状态查询按需加载。 return {"success": True, "data": _submission_data(record, include_images=False)} @app.get("/api/admin/answer-task-drafts") def admin_answer_task_drafts(_: AdminUser = Depends(current_user), db: Session = Depends(get_db)): records = db.scalars(select(AnswerSubmission).where(AnswerSubmission.status.in_(["待筛选", "待运营派发", "已派发", "待用户完成"])).order_by(AnswerSubmission.created_at.desc())).unique().all() return {"success": True, "data": [_submission_data(record, include_images=False) for record in records]} @app.get("/api/admin/answer-task-drafts/{submission_id}") def admin_answer_task_draft_detail(submission_id: int, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): record = db.get(AnswerSubmission, submission_id) if not record or record.status not in {"待筛选", "待运营派发", "已派发", "待用户完成"}: raise HTTPException(status_code=404, detail="回答任务不存在") return {"success": True, "data": _submission_data(record, include_images=True)} @app.post("/api/admin/answer-task-drafts/questions/{question_id}/operator-image") async def upload_answer_operator_images( question_id: int, files: list[UploadFile] = File(...), _: AdminUser = Depends(current_user), db: Session = Depends(get_db), ): """Attach up to three optional operation-specified images to one question.""" question = db.get(AnswerRecognizedQuestion, question_id) record = db.get(AnswerSubmission, question.submission_id) if question else None if not question or not record or record.status not in {"待筛选", "待运营派发", "已派发"}: raise HTTPException(status_code=404, detail="回答问题不存在或已经发布") if not files or len(files) > 3: raise HTTPException(status_code=400, detail="每道问题最多上传3张运营指定图片") old_ids = [value for value in (question.operator_image_screenshot_ids or "").split(",") if value.strip().isdigit()] for old_id in old_ids: old = db.get(AnswerScreenshot, int(old_id)) if old: db.delete(old) new_ids = [] for file in files: if file.content_type not in {"image/jpeg", "image/png", "image/webp"}: db.rollback() raise HTTPException(status_code=400, detail="运营指定图片仅支持 JPG、PNG、WEBP") image_data = await file.read() if not image_data: db.rollback() raise HTTPException(status_code=400, detail="不能上传空图片") image = AnswerScreenshot(submission_id=record.id, stage="operator", filename=file.filename or "operator-image", content_type=file.content_type, image_data=image_data) db.add(image); db.flush(); new_ids.append(str(image.id)) question.operator_image_screenshot_ids = ",".join(new_ids) question.operator_image_screenshot_id = int(new_ids[0]) if new_ids else None db.commit() db.refresh(record) return {"success": True, "data": _submission_data(record, include_images=True)} @app.post("/api/admin/answer-task-drafts/{submission_id}/reject") def reject_answer_task_draft(submission_id: int, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): record = db.get(AnswerSubmission, submission_id) if not record: raise HTTPException(status_code=404, detail="回答任务不存在") if record.status not in {"待筛选", "待运营派发"}: raise HTTPException(status_code=409, detail="当前任务不能打回") db.delete(record) db.commit() return {"success": True, "data": {"id": submission_id}} @app.post("/api/admin/ask-tasks/generate") def generate_ask_task(payload: dict, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): product_link = str(payload.get("product_link", "")).strip() product_name = str(payload.get("product_name", "")).strip() if not product_link: raise HTTPException(status_code=400, detail="商品链接为必填项") intent_error = _product_intent_error([product_name, payload.get("category"), payload.get("benefits"), payload.get("selling_points")]) if intent_error: raise HTTPException(status_code=400, detail=intent_error) options_error = _generation_options_error(payload, include_answers=False) if options_error: raise HTTPException(status_code=400, detail=options_error) try: count = int(payload.get("question_count", 5)) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="提问生成条数必须是正整数") if count < 1: raise HTTPException(status_code=400, detail="提问生成条数必须大于 0") raw_claim_limit = payload.get("claim_questions_per_user", "") if str(raw_claim_limit).strip(): try: claim_limit = int(raw_claim_limit) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="每个用户可认领的问题数量必须是 2-5 的整数") if claim_limit < 2 or claim_limit > 5: raise HTTPException(status_code=400, detail="每个用户可认领的问题数量必须在 2-5 条之间") else: claim_limit = 5 existing_task_id = payload.get("task_id") try: task = db.get(AskTask, int(existing_task_id)) if existing_task_id else None except (TypeError, ValueError): task = None if task: _require_ask_task_access(task, user) if task.status != "草稿": raise HTTPException(status_code=409, detail="已发布任务不能重新生成") task.product_link = product_link task.product_name = product_name or "待补充商品" task.category = str(payload.get("category", "")) task.benefits = str(payload.get("benefits", "")) task.selling_points = str(payload.get("selling_points", "")) task.language_style = str(payload.get("language_style", "")) task.target_group = str(payload.get("target_group", "")) task.usage_scene = str(payload.get("usage_scene", "")) task.purchase_channel = str(payload.get("purchase_channel", "")) task.claim_questions_per_user = claim_limit for question in list(task.questions): db.delete(question) db.flush() else: task = AskTask(task_no=f"TASK{datetime.utcnow().strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:4].upper()}", owner_id=user.id, product_link=product_link, product_name=product_name or "待补充商品", category=str(payload.get("category", "")), benefits=str(payload.get("benefits", "")), selling_points=str(payload.get("selling_points", "")), language_style=str(payload.get("language_style", "")), target_group=str(payload.get("target_group", "")), usage_scene=str(payload.get("usage_scene", "")), purchase_channel=str(payload.get("purchase_channel", "")), claim_questions_per_user=claim_limit, status="草稿") db.add(task); db.flush() questions = payload.get("questions") if isinstance(payload.get("questions"), list) else [] reference_text = str(payload.get("reference_qa", ""))[:20000] if not questions: questions = generate_questions(payload, count, reference_text) # 自定义问题优先，不足的数量用候选问题补齐，保证返回数量与运营填写值一致。 generated: list[str] = [] for value in questions: question = str(value).strip() if question and question not in generated: generated.append(question) if len(generated) == count: break if len(generated) < count: generated.extend(generate_questions(payload, count - len(generated), reference_text)) for index, question in enumerate(generated): db.add(AskQuestion(task_id=task.id, sequence_no=index + 1, question_text=question, selected=True)) db.commit(); db.refresh(task) return {"success": True, "data": _ask_task_data(task)} @app.patch("/api/admin/ask-tasks/{task_id}/questions/{question_id}") def update_ask_question(task_id: int, question_id: int, payload: dict, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): question = db.scalar(select(AskQuestion).where(AskQuestion.id == question_id, AskQuestion.task_id == task_id)) if not question or not _can_access_ask_task(question.task, user): raise HTTPException(status_code=404, detail="提问不存在") if question.task.status != "草稿": raise HTTPException(status_code=409, detail="已发布任务不能修改") if "question" in payload: value = str(payload.get("question", "")).strip() if not value: raise HTTPException(status_code=400, detail="提问内容不能为空") question.question_text = value if "selected" in payload: question.selected = bool(payload["selected"]) db.commit(); db.refresh(question) return {"success": True, "data": {"id": question.id, "question": question.question_text, "selected": question.selected}} @app.post("/api/admin/ask-tasks/{task_id}/questions") def add_ask_question(task_id: int, payload: dict, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): task = db.get(AskTask, task_id) text = str(payload.get("question", "")).strip() _require_ask_task_access(task, user) if task.status != "草稿": raise HTTPException(status_code=409, detail="已发布任务不能添加提问") if not text: raise HTTPException(status_code=400, detail="提问内容不能为空") question = AskQuestion(task_id=task.id, sequence_no=len(task.questions) + 1, question_text=text, selected=True) db.add(question); db.commit(); db.refresh(question) return {"success": True, "data": {"id": question.id, "sequence_no": question.sequence_no, "question": question.question_text, "selected": question.selected}} @app.post("/api/admin/ask-tasks/{task_id}/publish") def publish_ask_task(task_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): task = db.get(AskTask, task_id) _require_ask_task_access(task, user) if task.status != "草稿": raise HTTPException(status_code=409, detail="任务已经发布") if not any(q.selected for q in task.questions): raise HTTPException(status_code=400, detail="请至少选择一个提问") task.status = "已发布"; db.commit(); db.refresh(task) return {"success": True, "data": _ask_task_data(task)} @app.post("/api/admin/ask-tasks/{task_id}/cancel") def cancel_ask_task(task_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): task = db.get(AskTask, task_id) _require_ask_task_access(task, user) if task.status != "已发布": raise HTTPException(status_code=409, detail="当前任务不能撤销") task.status = "已撤销"; db.commit(); db.refresh(task) return {"success": True, "data": _ask_task_data(task)} @app.post("/api/admin/ask-tasks/{task_id}/complete") def complete_ask_task(task_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): task = db.get(AskTask, task_id) _require_ask_task_access(task, user) if task.status != "已发布": raise HTTPException(status_code=409, detail="当前任务不能完成归档") task.status = "已完成" task.updated_at = datetime.utcnow() db.commit() db.refresh(task) return {"success": True, "data": _ask_task_data(task)} @app.patch("/api/admin/answer-task-drafts/questions/{question_id}") def update_answer_question(question_id: int, payload: dict, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): question = db.get(AnswerRecognizedQuestion, question_id) if not question: raise HTTPException(status_code=404, detail="问题不存在") if "selected" in payload: question.selected = bool(payload["selected"]) if str(payload.get("answer", "")).strip(): question.answer_text = str(payload["answer"]).strip() if payload.get("regenerate"): question.answer_text = generate_answer_for_question(question.product_title, question.question_text) db.commit(); db.refresh(question) return {"success": True, "data": {"id": question.id, "answer": question.answer_text, "selected": question.selected}} @app.post("/api/admin/answer-task-drafts/{submission_id}/publish") def publish_answer_task(submission_id: int, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): record = db.get(AnswerSubmission, submission_id) if not record: raise HTTPException(status_code=404, detail="提交记录不存在") if record.status not in {"待筛选", "待运营派发", "已派发"}: raise HTTPException(status_code=409, detail="当前任务已经发布或不能发布") if not any(q.selected for q in record.questions): raise HTTPException(status_code=400, detail="请至少选择一个问题") record.status = "待用户完成"; db.commit() return {"success": True, "data": _submission_data(record, include_images=False)} @app.get("/api/admin/task-board") def admin_task_board(user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): if _release_expired_ask_assignments(db): db.commit() task_statuses = ["待用户完成", "待审核", "已通过", "已完成", "未通过", "已撤销"] records = db.scalars( select(AnswerSubmission).options( selectinload(AnswerSubmission.user), selectinload(AnswerSubmission.questions), noload(AnswerSubmission.screenshots), noload(AnswerSubmission.review_decisions), ) .where(AnswerSubmission.status.in_(task_statuses)) .order_by(AnswerSubmission.updated_at.desc()) ).unique().all() ask_statement = select(AskTask).options(selectinload(AskTask.owner)).where(AskTask.status.in_(["已发布", "已完成", "已撤销"]), _ask_task_owner_clause(user)).order_by(AskTask.updated_at.desc()) ask_tasks = db.scalars(ask_statement).unique().all() data = ( [_task_board_data(record, include_images=False) for record in records] + [_ask_task_data(task, include_questions=False) for task in ask_tasks] ) data.sort(key=lambda item: item.get("updated_at") or item.get("created_at") or "", reverse=True) return {"success": True, "data": data} @app.get("/api/admin/task-board/ask-tasks/{task_id}") def admin_task_board_ask_task_detail(task_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): task = db.get(AskTask, task_id) if not task or task.status not in {"已发布", "已完成", "已撤销"} or not _can_access_ask_task(task, user): raise HTTPException(status_code=404, detail="提问任务不存在") return {"success": True, "data": _ask_task_data(task, include_questions=True)} @app.delete("/api/admin/task-board/ask-tasks/{task_id}") def admin_task_board_delete_ask_task(task_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): task = db.get(AskTask, task_id) _require_ask_task_access(task, user) # AskTask's assignment relationship is intentionally not orphan-cascaded # because assignments may be created by users. Remove those dependent # records explicitly before deleting the task to keep every status # deletable without leaving invalid foreign keys behind. for assignment in list(task.assignments): db.delete(assignment) db.delete(task) db.commit() return {"success": True, "data": {"id": task_id}} @app.get("/api/admin/task-board/ask-batches/{batch_id}") def admin_task_board_ask_batch_detail(batch_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): batch = db.get(AskAssignmentBatch, batch_id) if not batch or batch.status not in {"approved", "paid"} or not _can_access_ask_task(batch.assignment.task, user): raise HTTPException(status_code=404, detail="待返款提问记录不存在") return {"success": True, "data": _ask_review_board_data(batch, include_images=True)} @app.delete("/api/admin/task-board/ask-batches/{batch_id}") def admin_task_board_delete_ask_batch(batch_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): batch = db.get(AskAssignmentBatch, batch_id) if not batch or not _can_access_ask_task(batch.assignment.task, user): raise HTTPException(status_code=404, detail="提问提交记录不存在") db.delete(batch) db.commit() return {"success": True, "data": {"id": batch_id}} @app.get("/api/admin/task-board/ask-assignments/{assignment_id}") def admin_task_board_ask_assignment_detail(assignment_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): assignment = db.get(AskAssignment, assignment_id) if not assignment or assignment.status not in {"已通过", "已完成"} or not _can_access_ask_task(assignment.task, user): raise HTTPException(status_code=404, detail="待返款提问记录不存在") return {"success": True, "data": _ask_review_board_data(assignment, include_images=True)} @app.delete("/api/admin/task-board/ask-assignments/{assignment_id}") def admin_task_board_delete_ask_assignment(assignment_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): assignment = db.get(AskAssignment, assignment_id) if not assignment or not _can_access_ask_task(assignment.task, user): raise HTTPException(status_code=404, detail="提问分配记录不存在") db.delete(assignment) db.commit() return {"success": True, "data": {"id": assignment_id}} @app.post("/api/admin/task-board/ask-batches/{batch_id}/confirm-refund") def confirm_ask_batch_refund(batch_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): batch = db.get(AskAssignmentBatch, batch_id) if not batch or not _can_access_ask_task(batch.assignment.task, user): raise HTTPException(status_code=404, detail="待返款提问记录不存在") if batch.status == "paid": raise HTTPException(status_code=409, detail="该任务已经完成返款") if batch.status != "approved": raise HTTPException(status_code=409, detail="只有审核通过的任务才能确认返款") batch.status = "paid" batch.updated_at = datetime.utcnow() db.commit(); db.refresh(batch) return {"success": True, "data": _ask_review_board_data(batch, include_images=False)} @app.post("/api/admin/task-board/ask-assignments/{assignment_id}/confirm-refund") def confirm_ask_assignment_refund(assignment_id: int, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): assignment = db.get(AskAssignment, assignment_id) if not assignment or not _can_access_ask_task(assignment.task, user): raise HTTPException(status_code=404, detail="待返款提问记录不存在") if assignment.status == "已完成": raise HTTPException(status_code=409, detail="该任务已经完成返款") if assignment.status != "已通过": raise HTTPException(status_code=409, detail="只有审核通过的任务才能确认返款") assignment.status = "已完成" assignment.updated_at = datetime.utcnow() db.commit(); db.refresh(assignment) return {"success": True, "data": _ask_review_board_data(assignment, include_images=False)} @app.get("/api/admin/task-board/{submission_id}") def admin_task_board_detail(submission_id: int, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): record = db.get(AnswerSubmission, submission_id) if not record or record.status not in {"待用户完成", "待审核", "已通过", "已完成", "未通过", "已撤销"}: raise HTTPException(status_code=404, detail="任务不存在") return {"success": True, "data": _task_board_data(record)} @app.delete("/api/admin/task-board/{submission_id}") def admin_task_board_delete_submission(submission_id: int, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): record = db.get(AnswerSubmission, submission_id) if not record: raise HTTPException(status_code=404, detail="任务不存在") db.delete(record) db.commit() return {"success": True, "data": {"id": submission_id}} @app.post("/api/admin/task-board/{submission_id}/confirm-refund") def confirm_answer_refund(submission_id: int, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): record = db.get(AnswerSubmission, submission_id) if not record: raise HTTPException(status_code=404, detail="待返款回答记录不存在") if record.status == "已完成": raise HTTPException(status_code=409, detail="该任务已经完成返款") if record.status != "已通过": raise HTTPException(status_code=409, detail="只有审核通过的任务才能确认返款") record.status = "已完成" db.commit(); db.refresh(record) return {"success": True, "data": _task_board_data(record, include_images=False)} @app.post("/api/admin/task-board/{submission_id}/cancel") def admin_task_board_cancel(submission_id: int, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): record = db.get(AnswerSubmission, submission_id) if not record: raise HTTPException(status_code=404, detail="任务不存在") if record.status == "已撤销": raise HTTPException(status_code=409, detail="任务已经撤销") if record.status == "已通过": raise HTTPException(status_code=409, detail="已完成任务不能撤销") if record.status not in {"待用户完成", "待审核", "未通过"}: raise HTTPException(status_code=400, detail="当前状态不能撤销") record.status = "已撤销" db.commit(); db.refresh(record) return {"success": True, "data": _task_board_data(record)} @app.post("/api/answer/me/submissions/{submission_id}/completion") async def answer_completion(submission_id: int, token: str = Form(...), files: list[UploadFile] = File(...), payment_files: list[UploadFile] = File(default=[]), db: Session = Depends(get_db)): user = _answer_user_from_token(token, db); record = db.get(AnswerSubmission, submission_id) if not record or record.user_id != user.id: raise HTTPException(status_code=404, detail="任务不存在") if record.status not in ("待用户完成", "未通过"): raise HTTPException(status_code=400, detail="当前任务不能提交完成截图") if not files or len(files) > 10: raise HTTPException(status_code=400, detail="最多上传10张截图") if not payment_files or len(payment_files) > 1: raise HTTPException(status_code=400, detail="请上传收款二维码后再提交") if record.status == "未通过": for old_screenshot in list(record.screenshots): if old_screenshot.stage in {"completion", "payment"}: db.delete(old_screenshot) for upload in files: if upload.content_type not in {"image/jpeg", "image/png", "image/webp"}: raise HTTPException(status_code=400, detail="仅支持图片") db.add(AnswerScreenshot(submission_id=record.id, stage="completion", filename=upload.filename or "completion", content_type=upload.content_type, image_data=await upload.read())) for upload in payment_files[:1]: if upload.content_type not in {"image/jpeg", "image/png", "image/webp"}: raise HTTPException(status_code=400, detail="收款码仅支持图片") db.add(AnswerScreenshot(submission_id=record.id, stage="payment", filename=upload.filename or "payment", content_type=upload.content_type, image_data=await upload.read())) record.status = "待审核"; db.commit() # 只返回轻量状态，避免重新编码截图阻塞审核状态切换。 return {"success": True, "data": _submission_data(record, include_images=False)} @app.get("/api/admin/answer-completion-reviews") def answer_completion_reviews(_: AdminUser = Depends(current_user), db: Session = Depends(get_db)): records = db.scalars( select(AnswerSubmission) .where(AnswerSubmission.status == "待审核") .order_by(AnswerSubmission.updated_at.desc()) ).unique().all() return {"success": True, "data": [_answer_review_data(record, include_images=False) for record in records]} @app.get("/api/admin/answer-completion-reviews/{submission_id}") def answer_completion_review_detail(submission_id: int, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): record = db.get(AnswerSubmission, submission_id) if not record: raise HTTPException(status_code=404, detail="任务不存在") return {"success": True, "data": _answer_review_data(record)} @app.post("/api/admin/answer-completion-reviews/{submission_id}/decision") def answer_completion_decision(submission_id: int, payload: dict, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): record = db.get(AnswerSubmission, submission_id) if not record: raise HTTPException(status_code=404, detail="任务不存在") if record.status != "待审核": raise HTTPException(status_code=409, detail="任务不在待审核状态") approved = bool(payload.get("approved")) reason_code = str(payload.get("reason_code", "")).strip() or None reason_text = str(payload.get("reason_text", "")).strip() or None if not approved and not (reason_code or reason_text): raise HTTPException(status_code=400, detail="请选择或填写审核不通过原因") db.add(AnswerReviewDecision(submission_id=record.id, approved=approved, reason_code=reason_code, reason_text=reason_text)) record.status = "已通过" if approved else "未通过"; db.commit(); db.refresh(record) return {"success": True, "data": _submission_data(record)} @app.post("/api/answer/public/submissions") async def public_answer_submission( client_id: str = Form(default="anonymous-browser"), user_name: str = Form(default=""), files: list[UploadFile] = File(...), ): if not files or len(files) > 10: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="最多上传10张截图") screenshots = [] for upload in files: if upload.content_type not in {"image/jpeg", "image/png", "image/webp"}: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 JPG、PNG、WEBP 图片") raw = await upload.read() screenshots.append({"name": upload.filename or "screenshot", "content_type": upload.content_type, "data_url": f"data:{upload.content_type};base64,{base64.b64encode(raw).decode()}"}) submission_id = f"ANS{datetime.utcnow().strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:4].upper()}" recognized = [{"id": f"Q{index + 1:03d}", "question": question["question"], "source_image": screenshots[index % len(screenshots)]["name"]} for index, question in enumerate(ANSWER_TASKS["TASK20260822002"]["questions"])] record = {"task_id": submission_id, "client_id": client_id.strip()[:64], "user_name": user_name.strip()[:64] or "未填写", "product_name": "待识别商品", "status": "待筛选", "recognized_questions": recognized, "invited_screenshots": screenshots, "final_screenshots": [], "selected_question_ids": [], "updated_at": datetime.utcnow().isoformat()} PUBLIC_ANSWER_SUBMISSIONS.append(record) return {"success": True, "data": record} @app.get("/api/answer/public/submissions") def public_answer_submission_history(client_id: str): if not client_id: return {"success": True, "data": []} records = [item for item in PUBLIC_ANSWER_SUBMISSIONS if item.get("client_id") == client_id] return {"success": True, "data": sorted(records, key=lambda item: item["updated_at"], reverse=True)} @app.get("/api/answer/tasks/{task_id}") def answer_task_detail(task_id: str, token: str): return {"success": True, "data": _public_answer_task(_answer_task(task_id, token))} @app.post("/api/answer/tasks/{task_id}/invited-screenshots") async def upload_invited_screenshots( task_id: str, token: str = Form(...), user_name: str = Form(default=""), files: list[UploadFile] = File(...), ): task = _answer_task(task_id, token) if not files or len(files) > 10: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="最多上传10张截图") screenshots = [] for upload in files: if upload.content_type not in {"image/jpeg", "image/png", "image/webp"}: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 JPG、PNG、WEBP 图片") raw = await upload.read() # MVP 暂存内存，后续接入对象存储；不写入数据库或磁盘。 screenshots.append({"name": upload.filename or "screenshot", "content_type": upload.content_type, "data_url": f"data:{upload.content_type};base64,{base64.b64encode(raw).decode()}"}) task["invited_screenshots"] = screenshots task["user_name"] = user_name.strip()[:64] task["recognized_questions"] = [ {"id": question["id"], "question": question["question"], "source_image": screenshots[index % len(screenshots)]["name"]} for index, question in enumerate(task["questions"]) ] task["status"] = "待筛选" task["updated_at"] = datetime.utcnow().isoformat() return {"success": True, "data": {"status": "已提交审核", "recognized_count": len(task["recognized_questions"])}} @app.post("/api/answer/tasks/{task_id}/select") def select_answer_questions(task_id: str, payload: dict): task = _answer_task(task_id, str(payload.get("token", ""))) question_ids = payload.get("question_ids") or [] valid_ids = {question["id"] for question in task["questions"]} task["selected_question_ids"] = [str(value) for value in question_ids if str(value) in valid_ids] task["status"] = "待完成" task["updated_at"] = datetime.utcnow().isoformat() return {"success": True, "data": _public_answer_task(task)} @app.post("/api/answer/tasks/{task_id}/final-screenshots") async def upload_final_screenshots( task_id: str, token: str = Form(...), files: list[UploadFile] = File(...), ): task = _answer_task(task_id, token) if not files or len(files) > 10: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="最多上传10张截图") screenshots = [] for upload in files: if upload.content_type not in {"image/jpeg", "image/png", "image/webp"}: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 JPG、PNG、WEBP 图片") raw = await upload.read() screenshots.append({"name": upload.filename or "screenshot", "content_type": upload.content_type, "data_url": f"data:{upload.content_type};base64,{base64.b64encode(raw).decode()}"}) task["final_screenshots"] = screenshots task["status"] = "待审核" task["updated_at"] = datetime.utcnow().isoformat() return {"success": True, "data": {"status": "已提交审核"}} @app.get("/api/admin/reviews") def reviews(_: AdminUser = Depends(current_user)): records = list(PUBLIC_ANSWER_SUBMISSIONS) for task in ANSWER_TASKS.values(): if task["status"] in {"待筛选", "待审核", "已通过", "未通过"}: records.append({ "task_id": task["task_id"], "user_name": task.get("user_name") or "未填写", "status": task["status"], "product_name": task["product_name"], "recognized_questions": task.get("recognized_questions", []), "selected_questions": [q for q in task["questions"] if q["id"] in set(task.get("selected_question_ids", []))], "invited_screenshots": [{"name": x["name"], "data_url": x["data_url"]} for x in task.get("invited_screenshots", [])], "final_screenshots": [{"name": x["name"], "data_url": x["data_url"]} for x in task.get("final_screenshots", [])], "updated_at": task["updated_at"], }) return {"success": True, "data": records} @app.post("/api/admin/reviews/{task_id}/decision") def review_decision(task_id: str, payload: dict, _: AdminUser = Depends(current_user)): task = ANSWER_TASKS.get(task_id) or next((item for item in PUBLIC_ANSWER_SUBMISSIONS if item["task_id"] == task_id), None) if not task: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="审核记录不存在") task["status"] = "已通过" if bool(payload.get("approved")) else "未通过" task["updated_at"] = datetime.utcnow().isoformat() return {"success": True, "data": {"task_id": task_id, "status": task["status"]}} @app.post("/api/admin/qa/generate") def qa_generate(payload: QAGenerateRequest, _: AdminUser = Depends(current_user)): intent_error = _product_intent_error([payload.product_name, payload.category, payload.benefits, payload.selling_points]) if intent_error: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=intent_error) options_error = _generation_options_error(payload.model_dump(), include_answers=True) if options_error: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=options_error) try: return {"success": True, "data": {"items": generate_qa(payload)}} except HTTPException: raise except Exception as exc: # Do not leak a raw 500 to the UI. Unexpected provider/parser errors # are surfaced as a retryable upstream error with a server traceback. import logging logging.getLogger(__name__).exception("QA generation failed") raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="问答生成服务异常，请检查大模型配置后重试") from exc def _qa_library_data(entry: QALibraryEntry) -> dict: try: answers = json.loads(entry.answers_json) except (TypeError, ValueError): answers = [entry.answers_json] if entry.answers_json else [] return { "id": entry.id, "product_name": entry.product_name, "category": entry.category, "benefits": entry.benefits, "selling_points": entry.selling_points, "question": entry.question_text, "answers": answers, "source": entry.source, "created_by": entry.created_by, "created_at": entry.created_at.isoformat(), } @app.get("/api/admin/qa-library") def qa_library(search: str = "", _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): statement = select(QALibraryEntry) keyword = search.strip() if keyword: pattern = f"%{keyword}%" statement = statement.where( or_( QALibraryEntry.product_name.like(pattern), QALibraryEntry.category.like(pattern), QALibraryEntry.question_text.like(pattern), QALibraryEntry.answers_json.like(pattern), ) ) records = db.scalars(statement.order_by(QALibraryEntry.created_at.desc()).limit(500)).all() return {"success": True, "data": [_qa_library_data(record) for record in records]} @app.post("/api/admin/qa-library") def save_qa_library(payload: dict, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): raw_items = payload.get("items") or [] if not isinstance(raw_items, list) or not raw_items: raise HTTPException(status_code=400, detail="请选择需要采纳的问答") product_name = str(payload.get("product_name", "")).strip() category = str(payload.get("category", "")).strip() benefits = str(payload.get("benefits", "")).strip() selling_points = str(payload.get("selling_points", "")).strip() prepared = [] for item in raw_items: question = str(item.get("question", "")).strip() answers = [str(value).strip() for value in (item.get("answers") or []) if str(value).strip()] if not question or not answers: continue normalized = json.dumps( {"product": product_name, "question": question, "answers": answers}, ensure_ascii=False, sort_keys=True, ) fingerprint = hashlib.sha256(normalized.encode("utf-8")).hexdigest() prepared.append((fingerprint, question, answers, str(item.get("source", "AI生成")).strip() or "AI生成")) if not prepared: raise HTTPException(status_code=400, detail="问答内容不能为空") fingerprints = [item[0] for item in prepared] existing = set( db.scalars(select(QALibraryEntry.fingerprint).where(QALibraryEntry.fingerprint.in_(fingerprints))).all() ) created = [] for fingerprint, question, answers, source in prepared: if fingerprint in existing: continue entry = QALibraryEntry( fingerprint=fingerprint, product_name=product_name, category=category, benefits=benefits, selling_points=selling_points, question_text=question, answers_json=json.dumps(answers, ensure_ascii=False), source=source, created_by=user.display_name, ) db.add(entry) created.append(entry) existing.add(fingerprint) db.commit() for entry in created: db.refresh(entry) return { "success": True, "data": { "added_count": len(created), "skipped_count": len(prepared) - len(created), "items": [_qa_library_data(entry) for entry in created], }, } @app.patch("/api/admin/qa-library/{entry_id}") def update_qa_library(entry_id: int, payload: dict, user: AdminUser = Depends(current_user), db: Session = Depends(get_db)): entry = db.get(QALibraryEntry, entry_id) if not entry: raise HTTPException(status_code=404, detail="问答记录不存在") question = str(payload.get("question", entry.question_text)).strip() answers = [str(value).strip() for value in (payload.get("answers") or []) if str(value).strip()] if not question or not answers: raise HTTPException(status_code=400, detail="问题和回答不能为空") product_name = str(payload.get("product_name", entry.product_name)).strip() category = str(payload.get("category", entry.category)).strip() benefits = str(payload.get("benefits", entry.benefits)).strip() selling_points = str(payload.get("selling_points", entry.selling_points)).strip() normalized = json.dumps( {"product": product_name, "question": question, "answers": answers}, ensure_ascii=False, sort_keys=True, ) fingerprint = hashlib.sha256(normalized.encode("utf-8")).hexdigest() duplicate = db.scalar( select(QALibraryEntry).where( QALibraryEntry.fingerprint == fingerprint, QALibraryEntry.id != entry.id, ) ) if duplicate: raise HTTPException(status_code=409, detail="修改后的内容已存在于问答库") entry.fingerprint = fingerprint entry.product_name = product_name entry.category = category entry.benefits = benefits entry.selling_points = selling_points entry.question_text = question entry.answers_json = json.dumps(answers, ensure_ascii=False) entry.created_by = user.display_name db.commit() db.refresh(entry) return {"success": True, "data": _qa_library_data(entry)} @app.delete("/api/admin/qa-library/{entry_id}") def delete_qa_library(entry_id: int, _: AdminUser = Depends(current_user), db: Session = Depends(get_db)): entry = db.get(QALibraryEntry, entry_id) if not entry: raise HTTPException(status_code=404, detail="问答记录不存在") db.delete(entry) db.commit() return {"success": True, "data": {"id": entry_id}} @app.post("/api/admin/auth/login") def login(payload: LoginRequest, request: Request, response: Response, db: Session = Depends(get_db)): user = db.scalar(select(AdminUser).where(AdminUser.username == payload.username)) valid = bool(user and user.is_active and verify_password(payload.password, user.password_hash)) db.add(AdminLoginLog(user_id=user.id if user else None, username=payload.username, success=valid, ip_address=request.client.host if request.client else None, user_agent=request.headers.get("user-agent"))) if not valid: db.commit() raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误") user.last_login_at = datetime.utcnow() db.commit() response.set_cookie(settings.cookie_name, create_session_token(user.id, user.role_code), httponly=True, secure=settings.cookie_secure, samesite="lax", max_age=settings.session_hours * 3600) return {"success": True, "data": {"user": public_user(user)}} @app.post("/api/admin/reference-qa/parse") async def parse_reference_qa(file: UploadFile = File(...), _: AdminUser = Depends(current_user)): """Extract readable question/answer text from an uploaded XLSX workbook. XLSX is a ZIP/XML format; parsing it on the server avoids treating the binary workbook as UTF-8 text in the browser (which produces mojibake). """ filename = (file.filename or "").lower() if not filename.endswith(".xlsx"): raise HTTPException(status_code=400, detail="仅支持解析 .xlsx 文件，请另存为 XLSX 后上传") raw = await file.read() try: try: from openpyxl import load_workbook workbook = load_workbook(io.BytesIO(raw), read_only=True, data_only=True) sheet = workbook.active rows = [[str(value).strip() if value is not None else "" for value in row] for row in sheet.iter_rows(values_only=True)] rows = [row for row in rows if any(row)] workbook.close() except ImportError: rows = None if rows is not None: if not rows: raise ValueError("工作表中没有可读取内容") header = [str(value).lower() for value in rows[0]] question_index = next((i for i, value in enumerate(header) if any(key in value for key in ("问题", "question", "q"))), None) answer_index = next((i for i, value in enumerate(header) if any(key in value for key in ("回答", "问答", "answer", "a"))), None) data_rows = rows[1:] if (question_index is not None or answer_index is not None) else rows lines = [] for values in data_rows: if question_index is not None and question_index < len(values): question = values[question_index] answer = values[answer_index] if answer_index is not None and answer_index < len(values) else "" if question or answer: lines.append(f"问题：{question}\n回答：{answer}") else: lines.append("\t".join(values)) text = "\n".join(lines).strip()[:20000] return {"success": True, "data": {"text": text, "filename": file.filename}} archive = zipfile.ZipFile(io.BytesIO(raw)) ns = {"x": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"} def xml_root(path: str): data = archive.read(path) # Some Windows exports declare UTF-8 but actually store Chinese # XML text as GB18030. Decode the XML bytes before parsing so the # original Chinese strings are not replaced with � characters. try: decoded = data.decode("utf-8") if "�" in decoded: decoded = data.decode("gb18030") except UnicodeDecodeError: decoded = data.decode("gb18030") return ET.fromstring(decoded) shared = [] if "xl/sharedStrings.xml" in archive.namelist(): root = xml_root("xl/sharedStrings.xml") shared = ["".join(node.itertext()) for node in root.findall("x:si", ns)] sheet_name = "xl/worksheets/sheet1.xml" if sheet_name not in archive.namelist(): raise ValueError("工作簿缺少首个工作表") root = xml_root(sheet_name) rows = [] for row in root.findall(".//x:sheetData/x:row", ns): values = [] for cell in row.findall("x:c", ns): cell_type = cell.attrib.get("t", "") value = cell.find("x:v", ns) if cell_type == "s" and value is not None: text = shared[int(value.text or "0")] if shared else "" elif cell_type == "inlineStr": text = "".join(cell.itertext()) else: text = value.text if value is not None and value.text else "" values.append(text.strip()) if any(values): rows.append(values) if not rows: raise ValueError("工作表中没有可读取内容") header = [str(value).lower() for value in rows[0]] question_index = next((i for i, value in enumerate(header) if any(key in value for key in ("问题", "question", "q"))), None) answer_index = next((i for i, value in enumerate(header) if any(key in value for key in ("回答", "问答", "answer", "a"))), None) lines = [] header = [str(value).lower() for value in rows[0]] question_index = next((i for i, value in enumerate(header) if any(key in value for key in ("问题", "question", "q"))), None) answer_index = next((i for i, value in enumerate(header) if any(key in value for key in ("回答", "问答", "answer", "a"))), None) data_rows = rows[1:] if (question_index is not None or answer_index is not None) else rows for values in data_rows: if question_index is not None and question_index < len(values): question = values[question_index] answer = values[answer_index] if answer_index is not None and answer_index < len(values) else "" if question or answer: lines.append(f"问题：{question}\n回答：{answer}") else: lines.append("\t".join(values)) text = "\n".join(lines).strip()[:20000] return {"success": True, "data": {"text": text, "filename": file.filename}} except (KeyError, ValueError, IndexError, ET.ParseError, zipfile.BadZipFile, UnicodeError) as exc: raise HTTPException(status_code=400, detail="无法读取该 XLSX 文件，请确认文件未损坏") from exc except Exception as exc: import logging logging.getLogger(__name__).exception("Reference XLSX parsing failed") raise HTTPException(status_code=400, detail="读取参考文件失败，请确认文件格式正确后重试") from exc @app.get("/api/admin/auth/dingtalk/config") def dingtalk_login_config(): return {"success": True, "data": {"enabled": _dingtalk_workbench_configured(), "corp_id": settings.dingtalk_corp_id, "agent_id": settings.dingtalk_agent_id, "redirect_uri": settings.dingtalk_redirect_uri}} @app.post("/api/admin/auth/dingtalk/workbench") def dingtalk_workbench_login(payload: dict, response: Response, db: Session = Depends(get_db)): if not _dingtalk_workbench_configured(): raise HTTPException(status_code=503, detail="钉钉工作台免登尚未配置，请联系管理员") code = str(payload.get("code") or "").strip() if not code: raise HTTPException(status_code=400, detail="钉钉免登授权码不能为空") profile = _dingtalk_workbench_user(code) user = _dingtalk_create_session(profile, db, response) return {"success": True, "data": {"user": public_user(user)}} @app.get("/api/admin/auth/dingtalk/start") def dingtalk_login_start(): if not _dingtalk_configured(): raise HTTPException(status_code=503, detail="钉钉登录尚未配置，请联系管理员") state = secrets.token_urlsafe(24) DINGTALK_STATES[state] = time.time() + 300 params = urllib.parse.urlencode({"redirect_uri": settings.dingtalk_redirect_uri, "client_id": settings.dingtalk_client_id, "response_type": "code", "scope": "openid", "state": state}) return {"success": True, "data": {"url": f"https://login.dingtalk.com/oauth2/auth?{params}"}} @app.get("/api/admin/auth/dingtalk/callback") def dingtalk_login_callback(response: Response, code: str = "", state: str = "", db: Session = Depends(get_db)): expires_at = DINGTALK_STATES.pop(state, 0) if not expires_at or expires_at < time.time(): raise HTTPException(status_code=400, detail="钉钉登录链接已失效，请重新从工作台打开") user = _dingtalk_login(code, db, response) response.headers["Location"] = f"{settings.frontend_url}/?dingtalk=success" response.status_code = 302 return response @app.post("/api/admin/auth/register", status_code=status.HTTP_201_CREATED) def register(_: RegisterRequest): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="运营后台已关闭自助注册，请联系管理员创建账号") @app.get("/api/admin/auth/me") def me(user: AdminUser = Depends(current_user)): return {"success": True, "data": {"user": public_user(user)}} @app.post("/api/admin/auth/logout") def logout(response: Response): response.delete_cookie(settings.cookie_name) return {"success": True} @app.get("/api/admin/auth/login-logs") def login_logs(_: AdminUser = Depends(require_super_admin), db: Session = Depends(get_db)): logs = db.scalars(select(AdminLoginLog).order_by(AdminLoginLog.created_at.desc()).limit(100)).all() return {"success": True, "data": [{"username": x.username, "success": x.success, "ip_address": x.ip_address, "created_at": x.created_at.isoformat()} for x in logs]} @app.get("/api/admin/users") def users(_: AdminUser = Depends(require_super_admin), db: Session = Depends(get_db)): records = db.scalars(select(AdminUser).order_by(AdminUser.id.asc())).all() return { "success": True, "data": [ { "id": user.id, "username": user.username, "display_name": user.display_name, "role_code": user.role_code, "role_label": ROLE_LABELS.get(user.role_code, user.role_code), "is_active": user.is_active, "last_login_at": user.last_login_at.isoformat() if user.last_login_at else None, } for user in records ], } @app.get("/api/admin/users/{user_id}") def user_detail(user_id: int, _: AdminUser = Depends(require_super_admin), db: Session = Depends(get_db)): user = db.get(AdminUser, user_id) if not user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在") return {"success": True, "data": public_user(user) | {"is_active": user.is_active, "last_login_at": user.last_login_at.isoformat() if user.last_login_at else None}} @app.post("/api/admin/users", status_code=status.HTTP_201_CREATED) def create_user(payload: AdminUserCreate, _: AdminUser = Depends(require_super_admin), db: Session = Depends(get_db)): if db.scalar(select(AdminUser).where(AdminUser.username == payload.username)): raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="用户名已存在") user = AdminUser(username=payload.username, password_hash=hash_password(payload.password), display_name=payload.display_name, role_code=payload.role_code) db.add(user) db.commit() db.refresh(user) return {"success": True, "data": {"user": public_user(user)}} @app.patch("/api/admin/users/{user_id}") def update_user(user_id: int, payload: AdminUserUpdate, current: AdminUser = Depends(require_super_admin), db: Session = Depends(get_db)): user = db.get(AdminUser, user_id) if not user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在") changes = payload.model_dump(exclude_unset=True) if user.id == current.id and (changes.get("is_active") is False or changes.get("role_code") == "operator"): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="不能停用或降级当前登录账号") if changes.get("role_code") == "operator" and user.role_code == "super_admin": super_count = db.scalar(select(func.count()).select_from(AdminUser).where(AdminUser.role_code == "super_admin", AdminUser.is_active.is_(True))) or 0 if super_count <= 1: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="至少需要保留一个启用中的超级管理员") if "password" in changes: user.password_hash = hash_password(changes.pop("password")) for key, value in changes.items(): setattr(user, key, value) db.commit() db.refresh(user) return {"success": True, "data": {"user": public_user(user)}} @app.delete("/api/admin/users/{user_id}") def delete_user(user_id: int, current: AdminUser = Depends(require_super_admin), db: Session = Depends(get_db)): user = db.get(AdminUser, user_id) if not user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在") if user.id == current.id: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="不能删除当前登录账号") if user.role_code == "super_admin": super_count = db.scalar(select(func.count()).select_from(AdminUser).where(AdminUser.role_code == "super_admin", AdminUser.is_active.is_(True))) or 0 if super_count <= 1: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="至少需要保留一个启用中的超级管理员") db.delete(user) db.commit() return {"success": True} @app.get("/api/admin/settings") def system_settings(_: AdminUser = Depends(require_super_admin)): return { "success": True, "data": { "app_name": settings.app_name, "session_hours": settings.session_hours, "database_driver": settings.database_url.split(":", 1)[0], "cookie_secure": settings.cookie_secure, "available_roles": [{"code": code, "label": label} for code, label in ROLE_LABELS.items()], }, }