65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Query
|
|
|
|
from app.schemas.common import ApiResponse
|
|
from app.services.domains_service import batch_update_domains, domain_filter_options, fetch_domain_detail, fetch_domains
|
|
|
|
router = APIRouter(tags=["domains"])
|
|
|
|
|
|
@router.get("/domains/filters", response_model=ApiResponse)
|
|
def domain_filters() -> ApiResponse:
|
|
return ApiResponse(data=domain_filter_options())
|
|
|
|
|
|
@router.get("/domains", response_model=ApiResponse)
|
|
def domain_list(
|
|
page: int = Query(default=1, ge=1),
|
|
page_size: int = Query(default=20, ge=1, le=200),
|
|
domain_keyword: str | None = Query(default=None),
|
|
register_status: int | None = Query(default=None),
|
|
detect_status: int | None = Query(default=None),
|
|
use_status: int | None = Query(default=None),
|
|
review_status: int | None = Query(default=None),
|
|
has_beian: int | None = Query(default=None),
|
|
beian_year: int | None = Query(default=None),
|
|
snapshot_year: str | None = Query(default=None),
|
|
website_url: str | None = Query(default=None),
|
|
company_type: str | None = Query(default=None),
|
|
source_type: int | None = Query(default=None),
|
|
backlink_gt_10: bool | None = Query(default=None),
|
|
) -> ApiResponse:
|
|
return ApiResponse(
|
|
data=fetch_domains(
|
|
page=page,
|
|
page_size=page_size,
|
|
domain_keyword=domain_keyword,
|
|
register_status=register_status,
|
|
detect_status=detect_status,
|
|
use_status=use_status,
|
|
review_status=review_status,
|
|
has_beian=has_beian,
|
|
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,
|
|
)
|
|
)
|
|
|
|
|
|
@router.get("/domains/{domain_id}", response_model=ApiResponse)
|
|
def domain_detail(domain_id: int) -> ApiResponse:
|
|
data = fetch_domain_detail(domain_id)
|
|
if not data:
|
|
return ApiResponse(code=1, message="域名不存在", data=None)
|
|
return ApiResponse(data=data)
|
|
|
|
|
|
@router.post("/domains/batch-update", response_model=ApiResponse)
|
|
def domain_batch_update(payload: dict) -> ApiResponse:
|
|
result = batch_update_domains(payload.get("domain_ids", []), payload.get("updates", {}))
|
|
return ApiResponse(message=f"成功更新 {result['updated_count']} 个域名", data=result)
|