207 lines
6.9 KiB
Python
207 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
from datetime import datetime
|
|
|
|
from openpyxl import Workbook
|
|
|
|
from app.core.db import get_db
|
|
from app.core.files import exports_root, load_export_records, save_export_record, timestamp_filename
|
|
from app.services.domains_service import (
|
|
BEIAN_STATUS_LABELS,
|
|
DETECT_STATUS_LABELS,
|
|
REGISTER_STATUS_LABELS,
|
|
REVIEW_STATUS_LABELS,
|
|
USE_STATUS_LABELS,
|
|
_build_domain_query_parts,
|
|
)
|
|
|
|
|
|
EXPORT_HEADERS = [
|
|
("domain", "域名"),
|
|
("register_status", "注册状态"),
|
|
("use_status", "使用状态"),
|
|
("detect_status", "检测状态"),
|
|
("review_status", "复核状态"),
|
|
("expire_date", "过期时间"),
|
|
("company_type", "单位性质"),
|
|
("has_beian", "备案状态"),
|
|
("website_url", "首页网址"),
|
|
("baidu_history", "百度历史收录"),
|
|
("baidu_site", "百度Site收录"),
|
|
("is_chinese_title", "标题为中文"),
|
|
("qihu360_site", "360 Site收录"),
|
|
("google_site", "Google Site收录"),
|
|
("beian_year", "备案年份"),
|
|
("snapshot_years", "快照年份"),
|
|
("backlink_count", "友链数"),
|
|
("backlink_gt_10", "友链>10"),
|
|
("detect_time", "检测时间"),
|
|
]
|
|
|
|
|
|
def _json_status_to_text(value) -> str:
|
|
if isinstance(value, dict):
|
|
status = value.get("status")
|
|
else:
|
|
status = None
|
|
return "是" if status else "否"
|
|
|
|
|
|
def _bool_to_text(value) -> str:
|
|
return "是" if bool(value) else "否"
|
|
|
|
|
|
def _normalize_payload(payload: dict) -> dict:
|
|
data = dict(payload or {})
|
|
data["page"] = int(data.get("page", 1) or 1)
|
|
data["page_size"] = int(data.get("page_size", 100) or 100)
|
|
data["page_count"] = int(data.get("page_count", 1) or 1)
|
|
data["scope"] = data.get("scope", "page")
|
|
data["type"] = data.get("type", "txt")
|
|
return data
|
|
|
|
|
|
def _query_export_rows(payload: dict) -> list[dict]:
|
|
data = _normalize_payload(payload)
|
|
from_clause, where_clause, params = _build_domain_query_parts(data)
|
|
|
|
limit_offset = ""
|
|
if data["scope"] == "page":
|
|
offset = (data["page"] - 1) * data["page_size"]
|
|
limit_offset = " limit %s offset %s"
|
|
params.extend([data["page_size"], offset])
|
|
elif data["scope"] == "pages":
|
|
offset = (data["page"] - 1) * data["page_size"]
|
|
limit_value = data["page_size"] * max(data["page_count"], 1)
|
|
limit_offset = " limit %s offset %s"
|
|
params.extend([limit_value, offset])
|
|
|
|
with get_db() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
f"""
|
|
select
|
|
d.domain,
|
|
d.register_status,
|
|
d.use_status,
|
|
d.detect_status,
|
|
d.review_status,
|
|
d.expire_date,
|
|
d.company_type,
|
|
d.has_beian,
|
|
d.website_url,
|
|
dd.baidu_history,
|
|
dd.baidu_site,
|
|
dd.is_chinese_title,
|
|
dd.qihu360_site,
|
|
dd.google_site,
|
|
d.beian_year,
|
|
d.snapshot_years,
|
|
d.backlink_count,
|
|
coalesce(dd.backlink_count_gt_10, false) as backlink_gt_10,
|
|
d.detect_time
|
|
{from_clause}
|
|
{where_clause}
|
|
order by d.id desc
|
|
{limit_offset}
|
|
""",
|
|
tuple(params),
|
|
)
|
|
rows = cur.fetchall()
|
|
|
|
result: list[dict] = []
|
|
for row in rows:
|
|
result.append(
|
|
{
|
|
"domain": row[0],
|
|
"register_status": REGISTER_STATUS_LABELS.get(row[1], str(row[1])),
|
|
"use_status": USE_STATUS_LABELS.get(row[2], str(row[2])),
|
|
"detect_status": DETECT_STATUS_LABELS.get(row[3], str(row[3])),
|
|
"review_status": REVIEW_STATUS_LABELS.get(row[4], str(row[4])),
|
|
"expire_date": row[5].isoformat(sep=" ", timespec="seconds") if row[5] else "",
|
|
"company_type": row[6] or "",
|
|
"has_beian": BEIAN_STATUS_LABELS.get(row[7], str(row[7])),
|
|
"website_url": row[8] or "",
|
|
"baidu_history": _json_status_to_text(row[9]),
|
|
"baidu_site": _json_status_to_text(row[10]),
|
|
"is_chinese_title": _bool_to_text(row[11]),
|
|
"qihu360_site": _json_status_to_text(row[12]),
|
|
"google_site": _json_status_to_text(row[13]),
|
|
"beian_year": row[14] or "",
|
|
"snapshot_years": row[15] or "",
|
|
"backlink_count": row[16] or 0,
|
|
"backlink_gt_10": "是" if row[17] else "否",
|
|
"detect_time": row[18].isoformat(sep=" ", timespec="seconds") if row[18] else "",
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def _write_txt(path, rows: list[dict]) -> None:
|
|
with path.open("w", encoding="utf-8") as handle:
|
|
for row in rows:
|
|
handle.write(f"{row['domain']}\n")
|
|
|
|
|
|
def _write_csv(path, rows: list[dict]) -> None:
|
|
# Add BOM so Excel on Chinese Windows opens CSV without mojibake.
|
|
with path.open("w", encoding="utf-8-sig", newline="") as handle:
|
|
writer = csv.writer(handle)
|
|
writer.writerow([label for _, label in EXPORT_HEADERS])
|
|
for row in rows:
|
|
writer.writerow([row[key] for key, _ in EXPORT_HEADERS])
|
|
|
|
|
|
def _write_xlsx(path, rows: list[dict]) -> None:
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "domains"
|
|
sheet.append([label for _, label in EXPORT_HEADERS])
|
|
for row in rows:
|
|
sheet.append([row[key] for key, _ in EXPORT_HEADERS])
|
|
workbook.save(path)
|
|
|
|
|
|
def create_export_file(payload: dict) -> dict:
|
|
data = _normalize_payload(payload)
|
|
rows = _query_export_rows(data)
|
|
|
|
ext = data["type"] if data["type"] in {"txt", "csv", "xlsx"} else "txt"
|
|
filename = timestamp_filename("domain_export", ext)
|
|
output_path = exports_root() / filename
|
|
|
|
if ext == "txt":
|
|
_write_txt(output_path, rows)
|
|
elif ext == "csv":
|
|
_write_csv(output_path, rows)
|
|
else:
|
|
_write_xlsx(output_path, rows)
|
|
|
|
created_at = datetime.fromtimestamp(output_path.stat().st_mtime)
|
|
record = {
|
|
"filename": filename,
|
|
"type": ext,
|
|
"scope": data["scope"],
|
|
"page": data["page"],
|
|
"page_size": data["page_size"],
|
|
"page_count": data["page_count"],
|
|
"count": len(rows),
|
|
"created_at": created_at.isoformat(sep=" ", timespec="seconds"),
|
|
"download_path": f"/api/v1/exports/download/{filename}",
|
|
}
|
|
save_export_record(record)
|
|
return record
|
|
|
|
|
|
def list_exports() -> list[dict]:
|
|
records = load_export_records()
|
|
normalized: list[dict] = []
|
|
for record in records:
|
|
item = dict(record)
|
|
created_at = item.get("created_at")
|
|
if isinstance(created_at, (int, float)):
|
|
item["created_at"] = datetime.fromtimestamp(created_at).isoformat(sep=" ", timespec="seconds")
|
|
normalized.append(item)
|
|
return normalized
|