File size: 3,749 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """企业数据查询 API。"""
from __future__ import annotations
from datetime import date
from typing import Literal
from ninja import Router
from apps.core.security.auth import JWTOrSessionAuth
from apps.enterprise_data.schemas import EnterpriseProvidersOut, EnterpriseQueryOut
from apps.enterprise_data.services import EnterpriseDataService
RiskType = Literal["周边风险", "预警提醒", "自身风险", "历史风险"]
router = Router(tags=["企业数据查询"], auth=JWTOrSessionAuth())
def _service() -> EnterpriseDataService:
return EnterpriseDataService()
@router.get("/providers", response=EnterpriseProvidersOut)
def list_providers(request, include_tools: bool = False) -> EnterpriseProvidersOut:
return EnterpriseProvidersOut(**_service().list_providers(include_tools=include_tools))
@router.get("/companies/search", response=EnterpriseQueryOut)
def search_companies(
request,
keyword: str,
provider: str | None = None,
include_raw: bool = False,
) -> EnterpriseQueryOut:
return EnterpriseQueryOut(
**_service().search_companies(keyword=keyword, provider=provider, include_raw=include_raw)
)
@router.get("/companies/{company_id}", response=EnterpriseQueryOut)
def get_company_profile(
request,
company_id: str,
provider: str | None = None,
include_raw: bool = False,
) -> EnterpriseQueryOut:
return EnterpriseQueryOut(
**_service().get_company_profile(company_id=company_id, provider=provider, include_raw=include_raw)
)
@router.get("/companies/{company_id}/risks", response=EnterpriseQueryOut)
def get_company_risks(
request,
company_id: str,
risk_type: RiskType = "自身风险",
provider: str | None = None,
include_raw: bool = False,
) -> EnterpriseQueryOut:
return EnterpriseQueryOut(
**_service().get_company_risks(
company_id=company_id,
risk_type=risk_type,
provider=provider,
include_raw=include_raw,
)
)
@router.get("/companies/{company_id}/shareholders", response=EnterpriseQueryOut)
def get_company_shareholders(
request,
company_id: str,
provider: str | None = None,
include_raw: bool = False,
) -> EnterpriseQueryOut:
return EnterpriseQueryOut(
**_service().get_company_shareholders(company_id=company_id, provider=provider, include_raw=include_raw)
)
@router.get("/companies/{company_id}/personnel", response=EnterpriseQueryOut)
def get_company_personnel(
request,
company_id: str,
provider: str | None = None,
include_raw: bool = False,
) -> EnterpriseQueryOut:
return EnterpriseQueryOut(
**_service().get_company_personnel(company_id=company_id, provider=provider, include_raw=include_raw)
)
@router.get("/personnel/{hcgid}", response=EnterpriseQueryOut)
def get_person_profile(
request,
hcgid: str,
provider: str | None = None,
include_raw: bool = False,
) -> EnterpriseQueryOut:
return EnterpriseQueryOut(**_service().get_person_profile(hcgid=hcgid, provider=provider, include_raw=include_raw))
@router.get("/biddings/search", response=EnterpriseQueryOut)
def search_bidding_info(
request,
keyword: str,
search_type: int = 1,
bid_type: int = 4,
start_date: date | None = None,
end_date: date | None = None,
provider: str | None = None,
include_raw: bool = False,
) -> EnterpriseQueryOut:
return EnterpriseQueryOut(
**_service().search_bidding_info(
keyword=keyword,
search_type=search_type,
bid_type=bid_type,
start_date=start_date,
end_date=end_date,
provider=provider,
include_raw=include_raw,
)
)
|