update 26.7.25
This commit is contained in:
Executable
+1
@@ -0,0 +1 @@
|
||||
"""Local utility scripts for market_api."""
|
||||
+445
@@ -0,0 +1,445 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
from collections.abc import Callable, Sequence
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.parse import urlencode
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
|
||||
# ===== 直接运行时主要改这里 =====
|
||||
API_BASE = "http://127.0.0.1:8000"
|
||||
OUTPUT_PATH: str | Path | None = None
|
||||
OUTPUT_DIR: str | Path | None = None
|
||||
YEAR = 2026
|
||||
MONTH = 5
|
||||
CT = "是"
|
||||
SCOPE = "行政区"
|
||||
SCOPE_VALUE = "崇明"
|
||||
TIMEOUT = 15.0
|
||||
# ================================
|
||||
|
||||
VALID_SCOPES = {"上海", "行政区", "属地"}
|
||||
VALID_CT_VALUES = {"是", "否"}
|
||||
|
||||
|
||||
class AnalysisTxtError(RuntimeError):
|
||||
"""Raised when analysis txt generation cannot continue."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ApiCall:
|
||||
path: str
|
||||
params: dict[str, object]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AnalysisTask:
|
||||
filename: str
|
||||
prompt: str
|
||||
calls: tuple[ApiCall, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RunConfig:
|
||||
api_base: str
|
||||
output_path: Path
|
||||
year: int
|
||||
month: int
|
||||
ct: str
|
||||
scope: str
|
||||
scope_value: str | None
|
||||
timeout: float
|
||||
|
||||
|
||||
Fetcher = Callable[[str, str, dict[str, object], float], object]
|
||||
|
||||
|
||||
def _normalize_month(month: int) -> int:
|
||||
if not 1 <= month <= 12:
|
||||
raise ValueError("month 必须在 1-12 之间")
|
||||
return month
|
||||
|
||||
|
||||
def _normalize_scope(scope: str, scope_value: str | None) -> tuple[str, str | None]:
|
||||
normalized_scope = scope.strip()
|
||||
if normalized_scope not in VALID_SCOPES:
|
||||
raise ValueError("scope 必须是 上海、行政区 或 属地")
|
||||
|
||||
normalized_value = scope_value.strip() if scope_value is not None else ""
|
||||
if normalized_scope == "上海":
|
||||
return normalized_scope, None
|
||||
|
||||
if not normalized_value:
|
||||
raise ValueError(f"scope 为{normalized_scope}时 scope_value 不能为空")
|
||||
return normalized_scope, normalized_value
|
||||
|
||||
|
||||
def _normalize_ct(ct: str) -> str:
|
||||
normalized_ct = ct.strip()
|
||||
if normalized_ct not in VALID_CT_VALUES:
|
||||
raise ValueError("ct 必须是 是 或 否")
|
||||
return normalized_ct
|
||||
|
||||
|
||||
def _scope_label(scope: str, scope_value: str | None) -> str:
|
||||
if scope == "上海":
|
||||
return "上海"
|
||||
return f"{scope}（{scope_value}）"
|
||||
|
||||
|
||||
def _safe_filename_segment(value: str) -> str:
|
||||
safe_value = re.sub(r'[\\/:*?"<>|]+', "_", value)
|
||||
safe_value = re.sub(r"\s+", "_", safe_value).strip("._ ")
|
||||
return safe_value or "未命名"
|
||||
|
||||
|
||||
def _base_params(
|
||||
*,
|
||||
year: int,
|
||||
month: int,
|
||||
scope: str,
|
||||
scope_value: str | None,
|
||||
) -> dict[str, object]:
|
||||
params: dict[str, object] = {
|
||||
"year": year,
|
||||
"month": month,
|
||||
"scope": scope,
|
||||
}
|
||||
if scope_value is not None:
|
||||
params["scope_value"] = scope_value
|
||||
return params
|
||||
|
||||
|
||||
def build_analysis_tasks(
|
||||
*,
|
||||
year: int,
|
||||
month: int,
|
||||
ct: str,
|
||||
scope: str,
|
||||
scope_value: str | None,
|
||||
) -> list[AnalysisTask]:
|
||||
normalized_month = _normalize_month(month)
|
||||
normalized_ct = _normalize_ct(ct)
|
||||
normalized_scope, normalized_scope_value = _normalize_scope(scope, scope_value)
|
||||
label = _scope_label(normalized_scope, normalized_scope_value)
|
||||
tender_params = _base_params(
|
||||
year=year,
|
||||
month=normalized_month,
|
||||
scope=normalized_scope,
|
||||
scope_value=normalized_scope_value,
|
||||
)
|
||||
award_params = {
|
||||
**tender_params,
|
||||
"ct": normalized_ct,
|
||||
}
|
||||
|
||||
return [
|
||||
AnalysisTask(
|
||||
filename="01_公开市场整体态势.txt",
|
||||
prompt=f"请基于以下数据分析{label}公开市场整体态势：",
|
||||
calls=(
|
||||
ApiCall("/tender/overall", dict(tender_params)),
|
||||
ApiCall("/tender/region", dict(tender_params)),
|
||||
),
|
||||
),
|
||||
AnalysisTask(
|
||||
filename="02_公开市场行业分布.txt",
|
||||
prompt=f"请基于以下数据分析{label}公开市场行业分布：",
|
||||
calls=(ApiCall("/tender/industry", dict(tender_params)),),
|
||||
),
|
||||
AnalysisTask(
|
||||
filename="03_公开市场单月动向.txt",
|
||||
prompt=f"请基于以下数据分析{label}公开市场单月动向：",
|
||||
calls=(ApiCall("/tender/overall", dict(tender_params)),),
|
||||
),
|
||||
AnalysisTask(
|
||||
filename="04_运营商市场整体态势.txt",
|
||||
prompt=f"请基于以下数据分析{label}运营商市场整体态势：",
|
||||
calls=(ApiCall("/award/overall", dict(award_params)),),
|
||||
),
|
||||
AnalysisTask(
|
||||
filename="05_运营商市场行业分布.txt",
|
||||
prompt=f"请基于以下数据分析{label}运营商市场行业分布：",
|
||||
calls=(ApiCall("/award/industry", dict(award_params)),),
|
||||
),
|
||||
AnalysisTask(
|
||||
filename="06_运营商市场单月动向.txt",
|
||||
prompt=f"请基于以下数据分析{label}运营商市场单月动向：",
|
||||
calls=(ApiCall("/award/overall", dict(award_params)),),
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def _request_url(api_base: str, path: str, params: dict[str, object]) -> str:
|
||||
query = urlencode(params)
|
||||
return f"{api_base.rstrip('/')}{path}?{query}"
|
||||
|
||||
|
||||
def fetch_json(
|
||||
api_base: str,
|
||||
path: str,
|
||||
params: dict[str, object],
|
||||
timeout: float,
|
||||
) -> object:
|
||||
url = _request_url(api_base, path, params)
|
||||
request = Request(url, headers={"Accept": "application/json"})
|
||||
try:
|
||||
with urlopen(request, timeout=timeout) as response:
|
||||
payload = response.read().decode("utf-8")
|
||||
except HTTPError as exc:
|
||||
return _http_error_payload(exc)
|
||||
except URLError as exc:
|
||||
raise AnalysisTxtError(f"无法连接 FastAPI：{url}，原因：{exc.reason}") from exc
|
||||
|
||||
try:
|
||||
return json.loads(payload)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise AnalysisTxtError(f"接口返回的内容不是 JSON：{url}") from exc
|
||||
|
||||
|
||||
def _http_error_payload(exc: HTTPError) -> dict[str, object]:
|
||||
body = exc.read().decode("utf-8", errors="replace")
|
||||
try:
|
||||
response_content: object = json.loads(body)
|
||||
except json.JSONDecodeError:
|
||||
response_content = body
|
||||
return {
|
||||
"HTTP状态码": exc.code,
|
||||
"错误信息": f"接口返回 HTTP {exc.code}",
|
||||
"响应内容": response_content,
|
||||
}
|
||||
|
||||
|
||||
def render_task(task: AnalysisTask, results: Sequence[object]) -> str:
|
||||
lines = [task.prompt, ""]
|
||||
for call, result in zip(task.calls, results, strict=True):
|
||||
lines.extend(
|
||||
[
|
||||
f"调用接口GET {call.path} 记录数据结果",
|
||||
"请求参数：",
|
||||
json.dumps(call.params, ensure_ascii=False, indent=2),
|
||||
"数据结果：",
|
||||
json.dumps(result, ensure_ascii=False, indent=2),
|
||||
"",
|
||||
]
|
||||
)
|
||||
return "\n".join(lines).rstrip() + "\n"
|
||||
|
||||
|
||||
def render_unified_txt(task_results: Sequence[tuple[AnalysisTask, Sequence[object]]]) -> str:
|
||||
return "\n\n".join(
|
||||
render_task(task, results).rstrip()
|
||||
for task, results in task_results
|
||||
) + "\n"
|
||||
|
||||
|
||||
def default_output_filename(
|
||||
*,
|
||||
year: int,
|
||||
month: int,
|
||||
scope: str,
|
||||
scope_value: str | None,
|
||||
) -> str:
|
||||
filename_parts = [
|
||||
str(year),
|
||||
str(month),
|
||||
_safe_filename_segment(scope),
|
||||
]
|
||||
if scope_value is not None and scope_value.strip():
|
||||
filename_parts.append(_safe_filename_segment(scope_value))
|
||||
return f"{'_'.join(filename_parts)}数据分析结果.txt"
|
||||
|
||||
|
||||
def default_output_path(
|
||||
*,
|
||||
year: int,
|
||||
month: int,
|
||||
scope: str,
|
||||
scope_value: str | None,
|
||||
) -> Path:
|
||||
return Path("outputs") / default_output_filename(
|
||||
year=year,
|
||||
month=month,
|
||||
scope=scope,
|
||||
scope_value=scope_value,
|
||||
)
|
||||
|
||||
|
||||
def default_output_dir(
|
||||
*,
|
||||
year: int,
|
||||
month: int,
|
||||
scope: str,
|
||||
scope_value: str | None,
|
||||
) -> Path:
|
||||
return default_output_path(
|
||||
year=year,
|
||||
month=month,
|
||||
scope=scope,
|
||||
scope_value=scope_value,
|
||||
).parent
|
||||
|
||||
|
||||
def generate_files(
|
||||
*,
|
||||
api_base: str,
|
||||
year: int,
|
||||
month: int,
|
||||
ct: str,
|
||||
scope: str,
|
||||
scope_value: str | None,
|
||||
timeout: float,
|
||||
output_path: Path | None = None,
|
||||
output_dir: Path | None = None,
|
||||
fetcher: Fetcher = fetch_json,
|
||||
) -> list[Path]:
|
||||
tasks = build_analysis_tasks(
|
||||
year=year,
|
||||
month=month,
|
||||
ct=ct,
|
||||
scope=scope,
|
||||
scope_value=scope_value,
|
||||
)
|
||||
if output_path is None:
|
||||
if output_dir is None:
|
||||
output_path = default_output_path(
|
||||
year=year,
|
||||
month=month,
|
||||
scope=scope,
|
||||
scope_value=scope_value,
|
||||
)
|
||||
else:
|
||||
output_path = output_dir / default_output_filename(
|
||||
year=year,
|
||||
month=month,
|
||||
scope=scope,
|
||||
scope_value=scope_value,
|
||||
)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
task_results: list[tuple[AnalysisTask, Sequence[object]]] = []
|
||||
for task in tasks:
|
||||
results = [
|
||||
fetcher(api_base, call.path, call.params, timeout)
|
||||
for call in task.calls
|
||||
]
|
||||
task_results.append((task, results))
|
||||
|
||||
output_path.write_text(render_unified_txt(task_results), encoding="utf-8")
|
||||
written_files = [output_path]
|
||||
return written_files
|
||||
|
||||
|
||||
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="调用 marketAPI 并生成数据分析基础 txt 文件。",
|
||||
)
|
||||
parser.add_argument("--api-base", help="覆盖脚本顶部的 API_BASE。")
|
||||
parser.add_argument("--output-path", type=Path, help="覆盖脚本顶部的 OUTPUT_PATH。")
|
||||
parser.add_argument(
|
||||
"--output-dir",
|
||||
type=Path,
|
||||
help="自定义输出目录，文件名按规则生成。",
|
||||
)
|
||||
parser.add_argument("--year", type=int, help="覆盖脚本顶部的 YEAR。")
|
||||
parser.add_argument("--month", type=int, help="覆盖脚本顶部的 MONTH。")
|
||||
parser.add_argument("--ct", help="覆盖脚本顶部的 CT。")
|
||||
parser.add_argument("--scope", help="覆盖脚本顶部的 SCOPE。")
|
||||
parser.add_argument("--scope-value", help="覆盖脚本顶部的 SCOPE_VALUE。")
|
||||
parser.add_argument(
|
||||
"--timeout",
|
||||
type=float,
|
||||
help="覆盖脚本顶部的 TIMEOUT。",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def _output_path_from_config(
|
||||
output_path: str | Path | None,
|
||||
output_dir: str | Path | None,
|
||||
*,
|
||||
year: int,
|
||||
month: int,
|
||||
scope: str,
|
||||
scope_value: str | None,
|
||||
) -> Path:
|
||||
if output_path is not None and str(output_path).strip():
|
||||
return Path(output_path)
|
||||
|
||||
if output_dir is not None and str(output_dir).strip():
|
||||
return Path(output_dir) / default_output_filename(
|
||||
year=year,
|
||||
month=month,
|
||||
scope=scope,
|
||||
scope_value=scope_value,
|
||||
)
|
||||
|
||||
return default_output_path(
|
||||
year=year,
|
||||
month=month,
|
||||
scope=scope,
|
||||
scope_value=scope_value,
|
||||
)
|
||||
|
||||
|
||||
def resolve_run_config(args: argparse.Namespace) -> RunConfig:
|
||||
year = args.year if args.year is not None else YEAR
|
||||
month = _normalize_month(args.month if args.month is not None else MONTH)
|
||||
ct = _normalize_ct(args.ct if args.ct is not None else CT)
|
||||
scope_input = args.scope if args.scope is not None else SCOPE
|
||||
scope_value_input = args.scope_value if args.scope_value is not None else SCOPE_VALUE
|
||||
output_path_input = args.output_path if args.output_path is not None else OUTPUT_PATH
|
||||
output_dir_input = args.output_dir if args.output_dir is not None else OUTPUT_DIR
|
||||
timeout = args.timeout if args.timeout is not None else TIMEOUT
|
||||
api_base = args.api_base if args.api_base is not None else API_BASE
|
||||
|
||||
scope = scope_input.strip()
|
||||
scope, scope_value = _normalize_scope(scope, scope_value_input)
|
||||
output_path = _output_path_from_config(
|
||||
output_path_input,
|
||||
output_dir_input,
|
||||
year=year,
|
||||
month=month,
|
||||
scope=scope,
|
||||
scope_value=scope_value,
|
||||
)
|
||||
|
||||
return RunConfig(
|
||||
api_base=api_base,
|
||||
output_path=output_path,
|
||||
year=year,
|
||||
month=month,
|
||||
ct=ct,
|
||||
scope=scope,
|
||||
scope_value=scope_value,
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
config = resolve_run_config(parse_args(argv))
|
||||
|
||||
written_files = generate_files(
|
||||
api_base=config.api_base,
|
||||
output_path=config.output_path,
|
||||
year=config.year,
|
||||
month=config.month,
|
||||
ct=config.ct,
|
||||
scope=config.scope,
|
||||
scope_value=config.scope_value,
|
||||
timeout=config.timeout,
|
||||
)
|
||||
|
||||
print(f"已生成 {len(written_files)} 个 txt 文件：")
|
||||
for path in written_files:
|
||||
print(path)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user