from __future__ import annotations from datetime import datetime from math import ceil from app.core.db import get_db DETECT_STATUS_LABELS = { 0: "待检测", 1: "检测通过", 2: "检测中", 3: "黑名单", 4: "检测失败", } REGISTER_STATUS_LABELS = { 0: "待检测", 2: "可注册", 3: "已注册", 4: "宽限期", 5: "赎回期", 6: "删除期", 7: "clientHold", 8: "serverHold", 9: "状态未知", 10: "检测失败", } USE_STATUS_LABELS = { 0: "未使用", 1: "已经使用", 2: "已经卖出", 3: "已经预定", } REVIEW_STATUS_LABELS = { 0: "无需复核", 1: "待人工复核", 2: "人工通过", 3: "人工拒绝", } SOURCE_TYPE_LABELS = { 1: "聚名一口价", 2: "聚名过期删除", 6: "手工录入", 7: "TXT 导入", 9: "其它", } BEIAN_STATUS_LABELS = { 1: "未检测", 2: "有备案", 3: "无备案", } def _format_timestamp(value) -> str | None: if not value: return None return value.isoformat(sep=" ", timespec="seconds") def _json_status_to_text(value) -> str: if isinstance(value, dict): status = value.get("status") else: status = None return "是" if status else "否" def _json_state(value) -> str: if isinstance(value, dict): return str(value.get("state") or "").strip() return "" def _json_message(value) -> str: if isinstance(value, dict): return str(value.get("message") or "").strip() return "" def _normalize_step_detail(label: str, value) -> dict: payload = value if isinstance(value, dict) else {} state = str(payload.get("state") or "").strip() return { "label": label, "state": state or ("passed" if bool(payload.get("status")) else ""), "status": bool(payload.get("status")) if isinstance(payload.get("status"), bool) else None, "message": str(payload.get("message") or "").strip(), "checked_at": str(payload.get("checked_at") or "").strip(), "step": str(payload.get("step") or "").strip(), "raw": payload, } def _build_step_details(row: tuple) -> list[dict]: return [ _normalize_step_detail("百度历史收录", row[16]), _normalize_step_detail("百度Site收录", row[17]), { "label": "标题为中文", "state": "passed" if bool(row[18]) else "", "status": bool(row[18]), "message": "标题含中文" if bool(row[18]) else "", "checked_at": "", "step": "中文标题", "raw": row[18], }, _normalize_step_detail("360 Site收录", row[19]), _normalize_step_detail("Google Site收录", row[20]), _normalize_step_detail("时光机", row[21]), _normalize_step_detail("站长之家", row[22]), _normalize_step_detail("爱站网", row[23]), _normalize_step_detail("桔子SEO", row[24]), _normalize_step_detail("聚查", row[25]), ] def _summarize_step_details(step_details: list[dict]) -> dict: degraded = [item for item in step_details if item.get("state") == "degraded"] failed = [item for item in step_details if item.get("state") == "failed"] blacklisted = [item for item in step_details if item.get("state") == "blacklisted"] return { "degraded_count": len(degraded), "failed_count": len(failed), "blacklisted_count": len(blacklisted), "has_degraded": bool(degraded), "has_failed": bool(failed), "has_blacklisted_step": bool(blacklisted), "summary_text": ( f"降级 {len(degraded)} / 失败 {len(failed)} / 命中 {len(blacklisted)}" if degraded or failed or blacklisted else "步骤正常" ), } def _bool_to_text(value) -> str: return "是" if bool(value) else "否" def _normalize_detection_update(value) -> bool | None: if value in (None, "", "skip"): return None if isinstance(value, bool): return value text = str(value).strip().lower() if text in {"是", "true", "1", "yes"}: return True if text in {"否", "false", "0", "no"}: return False raise ValueError("检测结果字段仅支持“是”或“否”") def _now_text() -> str: return datetime.now().isoformat(sep=" ", timespec="seconds") def _merge_detection_step_update(field: str, value: bool, existing_payload) -> dict | bool: if field == "is_chinese_title": return bool(value) existing = dict(existing_payload or {}) if isinstance(existing_payload, dict) else {} previous_status = existing.get("status") if isinstance(existing.get("status"), bool) else None merged = dict(existing) merged["status"] = bool(value) merged["state"] = "passed" if bool(value) else "failed" merged["step"] = str(merged.get("step") or field) if previous_status != bool(value) or not str(merged.get("message") or "").strip(): merged["message"] = "人工批量更新" if previous_status != bool(value) or not str(merged.get("checked_at") or "").strip(): merged["checked_at"] = _now_text() merged["manual_override"] = True return merged def _build_domain_query_parts(filters: dict | None = None) -> tuple[str, str, list[object]]: filters = filters or {} conditions: list[str] = [] params: list[object] = [] if filters.get("domain_keyword"): conditions.append("d.domain ilike %s") params.append(f"%{str(filters['domain_keyword']).strip()}%") if filters.get("register_status") is not None: conditions.append("d.register_status = %s") params.append(filters["register_status"]) if filters.get("detect_status") is not None: conditions.append("d.detect_status = %s") params.append(filters["detect_status"]) if filters.get("use_status") is not None: conditions.append("d.use_status = %s") params.append(filters["use_status"]) if filters.get("review_status") is not None: conditions.append("d.review_status = %s") params.append(filters["review_status"]) if filters.get("has_beian") is not None: conditions.append("d.has_beian = %s") params.append(filters["has_beian"]) if filters.get("beian_year"): conditions.append("d.beian_year = %s") params.append(int(filters["beian_year"])) if filters.get("snapshot_year"): conditions.append("coalesce(d.snapshot_years, '') like %s") params.append(f"%{str(filters['snapshot_year']).strip()}%") if filters.get("website_url"): conditions.append("coalesce(d.website_url, '') ilike %s") params.append(f"%{str(filters['website_url']).strip()}%") if filters.get("company_type"): conditions.append("coalesce(d.company_type, '') = %s") params.append(str(filters["company_type"]).strip()) if filters.get("source_type") is not None: conditions.append("d.source_type = %s") params.append(int(filters["source_type"])) if filters.get("backlink_gt_10") is not None: conditions.append("coalesce(dd.backlink_count_gt_10, false) = %s") params.append(bool(filters["backlink_gt_10"])) from_clause = """ from domains d left join domain_detections dd on dd.domain_id = d.id """ where_clause = f"where {' and '.join(conditions)}" if conditions else "" return from_clause, where_clause, params def fetch_domains( page: int = 1, page_size: int = 20, domain_keyword: str | None = None, register_status: int | None = None, detect_status: int | None = None, has_beian: int | None = None, use_status: int | None = None, review_status: int | None = None, beian_year: int | None = None, snapshot_year: str | None = None, website_url: str | None = None, company_type: str | None = None, source_type: int | None = None, backlink_gt_10: bool | None = None, ) -> dict: offset = (page - 1) * page_size filters = { "domain_keyword": domain_keyword, "register_status": register_status, "detect_status": detect_status, "has_beian": has_beian, "use_status": use_status, "review_status": review_status, "beian_year": beian_year, "snapshot_year": snapshot_year, "website_url": website_url, "company_type": company_type, "source_type": source_type, "backlink_gt_10": backlink_gt_10, } from_clause, where_clause, params = _build_domain_query_parts(filters) with get_db() as conn: with conn.cursor() as cur: cur.execute(f"select count(*) {from_clause} {where_clause}", tuple(params)) total = cur.fetchone()[0] cur.execute( f""" select d.id, d.domain, d.register_status, d.use_status, d.detect_status, d.review_status, d.has_beian, d.website_url, d.beian_year, d.snapshot_years, d.backlink_count, d.detect_time, d.source_type, coalesce(dd.backlink_count_gt_10, false) as backlink_gt_10, d.expire_date, d.company_type, dd.baidu_history, dd.baidu_site, dd.is_chinese_title, dd.qihu360_site, dd.google_site, dd.wayback_info, dd.chinaz_info, dd.aizhan_info, dd.juziseo_info, dd.jucha_info {from_clause} {where_clause} order by d.id desc limit %s offset %s """, tuple(params + [page_size, offset]), ) rows = cur.fetchall() items = [] for row in rows: step_details = _build_step_details(row) step_summary = _summarize_step_details(step_details) items.append({ "id": row[0], "domain": row[1], "register_status": REGISTER_STATUS_LABELS.get(row[2], str(row[2])), "register_status_code": row[2], "use_status": USE_STATUS_LABELS.get(row[3], str(row[3])), "use_status_code": row[3], "detect_status": DETECT_STATUS_LABELS.get(row[4], str(row[4])), "detect_status_code": row[4], "review_status": REVIEW_STATUS_LABELS.get(row[5], str(row[5])), "review_status_code": row[5], "has_beian": BEIAN_STATUS_LABELS.get(row[6], str(row[6])), "has_beian_code": row[6], "website_url": row[7] or "", "beian_year": row[8], "snapshot_years": row[9] or "", "backlink_count": row[10], "detect_time": _format_timestamp(row[11]), "source_type": row[12], "source_label": SOURCE_TYPE_LABELS.get(row[12], str(row[12])), "backlink_gt_10": row[13], "expire_date": _format_timestamp(row[14]), "company_type": row[15] or "", "baidu_history": _json_status_to_text(row[16]), "baidu_site": _json_status_to_text(row[17]), "is_chinese_title": _bool_to_text(row[18]), "qihu360_site": _json_status_to_text(row[19]), "google_site": _json_status_to_text(row[20]), "step_summary": step_summary, }) return { "list": items, "page": page, "page_size": page_size, "total": total, "pages": ceil(total / page_size) if page_size else 1, } def fetch_domain_detail(domain_id: int) -> dict | None: with get_db() as conn: with conn.cursor() as cur: cur.execute( """ select d.id, d.domain, d.register_status, d.use_status, d.detect_status, d.review_status, d.has_beian, d.website_url, d.beian_year, d.snapshot_years, d.backlink_count, d.detect_time, d.source_type, coalesce(dd.backlink_count_gt_10, false) as backlink_gt_10, d.expire_date, d.company_type, dd.baidu_history, dd.baidu_site, dd.is_chinese_title, dd.qihu360_site, dd.google_site, dd.wayback_info, dd.chinaz_info, dd.aizhan_info, dd.juziseo_info, dd.jucha_info from domains d left join domain_detections dd on dd.domain_id = d.id where d.id = %s limit 1 """, (int(domain_id),), ) row = cur.fetchone() if not row: return None step_details = _build_step_details(row) step_summary = _summarize_step_details(step_details) return { "id": row[0], "domain": row[1], "register_status": REGISTER_STATUS_LABELS.get(row[2], str(row[2])), "register_status_code": row[2], "use_status": USE_STATUS_LABELS.get(row[3], str(row[3])), "use_status_code": row[3], "detect_status": DETECT_STATUS_LABELS.get(row[4], str(row[4])), "detect_status_code": row[4], "review_status": REVIEW_STATUS_LABELS.get(row[5], str(row[5])), "review_status_code": row[5], "has_beian": BEIAN_STATUS_LABELS.get(row[6], str(row[6])), "has_beian_code": row[6], "website_url": row[7] or "", "beian_year": row[8], "snapshot_years": row[9] or "", "backlink_count": row[10], "detect_time": _format_timestamp(row[11]), "source_type": row[12], "source_label": SOURCE_TYPE_LABELS.get(row[12], str(row[12])), "backlink_gt_10": row[13], "expire_date": _format_timestamp(row[14]), "company_type": row[15] or "", "step_summary": step_summary, "step_details": step_details, "raw_detection": { "baidu_history": row[16], "baidu_site": row[17], "is_chinese_title": row[18], "qihu360_site": row[19], "google_site": row[20], "wayback_info": row[21], "chinaz_info": row[22], "aizhan_info": row[23], "juziseo_info": row[24], "jucha_info": row[25], }, } def domain_filter_options() -> dict: return { "register_status": [ {"label": label, "value": value} for value, label in REGISTER_STATUS_LABELS.items() if value in (0, 2, 3, 4, 5, 6, 7, 8, 9, 10) ], "detect_status": [ {"label": label, "value": value} for value, label in DETECT_STATUS_LABELS.items() ], "use_status": [ {"label": label, "value": value} for value, label in USE_STATUS_LABELS.items() ], "review_status": [ {"label": label, "value": value} for value, label in REVIEW_STATUS_LABELS.items() ], "has_beian": [ {"label": "未检测", "value": 1}, {"label": "有备案", "value": 2}, {"label": "无备案", "value": 3}, ], "source_type": [ {"label": label, "value": value} for value, label in SOURCE_TYPE_LABELS.items() ], "supports_backlink_gt_10": True, "supports_txt_export": True, "supports_excel_export": True, "supports_multi_page_export": True, } def batch_update_domains(domain_ids: list[int], updates: dict) -> dict: if not domain_ids: raise ValueError("未选择需要更新的域名") allowed_fields = { "review_status", "expire_date", "has_beian", "beian_year", "snapshot_years", "company_type", "detect_time", "website_url", "backlink_count", "baidu_history", "baidu_site", "is_chinese_title", "qihu360_site", "google_site", } payload = {key: value for key, value in updates.items() if key in allowed_fields and value not in (None, "", "skip")} if not payload: raise ValueError("没有可更新的字段") domain_fields = { "review_status", "expire_date", "has_beian", "beian_year", "snapshot_years", "company_type", "detect_time", "website_url", "backlink_count", } detection_fields = { "baidu_history", "baidu_site", "is_chinese_title", "qihu360_site", "google_site", } updated_count = 0 with get_db() as conn: with conn.cursor() as cur: for domain_id in domain_ids: set_parts: list[str] = [] params: list[object] = [] for field, value in payload.items(): if field not in domain_fields: continue if field == "backlink_count": set_parts.append("backlink_count = %s") params.append(int(value)) else: set_parts.append(f"{field} = %s") params.append(value) if set_parts: params.append(domain_id) cur.execute( f"update domains set {', '.join(set_parts)}, update_time = now() where id = %s", tuple(params), ) existing_detection_payloads: dict[str, object] = {} existing_detection_row = None if "backlink_count" in payload or detection_fields.intersection(payload.keys()): cur.execute( """ select id, baidu_history, baidu_site, is_chinese_title, qihu360_site, google_site, backlink_count_gt_10 from domain_detections where domain_id = %s """, (domain_id,), ) existing_detection_row = cur.fetchone() if existing_detection_row: existing_detection_payloads = { "baidu_history": existing_detection_row[1], "baidu_site": existing_detection_row[2], "is_chinese_title": existing_detection_row[3], "qihu360_site": existing_detection_row[4], "google_site": existing_detection_row[5], } detection_payload: dict[str, object] = {} for field in detection_fields: if field not in payload: continue normalized = _normalize_detection_update(payload[field]) if normalized is None: continue detection_payload[field] = _merge_detection_step_update( field, normalized, existing_detection_payloads.get(field), ) if "backlink_count" in payload: backlink_gt_10 = int(payload["backlink_count"]) > 10 if existing_detection_row: cur.execute( "update domain_detections set backlink_count_gt_10 = %s, update_time = now() where domain_id = %s", (backlink_gt_10, domain_id), ) else: cur.execute( """ insert into domain_detections (domain_id, backlink_count_gt_10, create_time, update_time) values (%s, %s, now(), now()) """, (domain_id, backlink_gt_10), ) if detection_payload: if existing_detection_row: detection_set_parts: list[str] = [] detection_params: list[object] = [] for field, value in detection_payload.items(): detection_set_parts.append(f"{field} = %s") detection_params.append(value) detection_params.append(domain_id) cur.execute( f""" update domain_detections set {', '.join(detection_set_parts)}, update_time = now() where domain_id = %s """, tuple(detection_params), ) else: insert_fields = ["domain_id", *detection_payload.keys(), "create_time", "update_time"] placeholders = ["%s"] * (1 + len(detection_payload)) + ["now()", "now()"] insert_params = [domain_id, *detection_payload.values()] cur.execute( f""" insert into domain_detections ({', '.join(insert_fields)}) values ({', '.join(placeholders)}) """, tuple(insert_params), ) updated_count += 1 conn.commit() return { "updated_count": updated_count, "fields": sorted(payload.keys()), }