265 lines
8.5 KiB
Python
265 lines
8.5 KiB
Python
"""
|
||
Public GraphQL Schema for company data.
|
||
|
||
This endpoint provides:
|
||
- companyTeaser: public data (no auth required)
|
||
- companyFull: full data (requires authentication)
|
||
|
||
Data is read from MongoDB company_documents collection.
|
||
Queries can be by INN (direct) or by KYC Profile UUID.
|
||
"""
|
||
|
||
import graphene
|
||
from django.conf import settings
|
||
from pymongo import MongoClient
|
||
|
||
from ..models import KYCProfile, KYCDetailsRussia
|
||
|
||
|
||
def get_mongo_client():
|
||
"""Get MongoDB client."""
|
||
if not settings.MONGODB_URI:
|
||
return None
|
||
return MongoClient(settings.MONGODB_URI)
|
||
|
||
|
||
def get_company_documents(inn: str) -> list:
|
||
"""Get all documents for a company by INN."""
|
||
client = get_mongo_client()
|
||
if not client:
|
||
return []
|
||
|
||
try:
|
||
db = client[settings.MONGODB_DB]
|
||
collection = db["company_documents"]
|
||
return list(collection.find({"inn": inn}))
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def get_inn_by_profile_uuid(profile_uuid: str) -> str | None:
|
||
"""Get INN from KYCProfile by its UUID."""
|
||
try:
|
||
profile = KYCProfile.objects.get(uuid=profile_uuid)
|
||
# Get country details (KYCDetailsRussia) via GenericForeignKey
|
||
if profile.country_details and isinstance(profile.country_details, KYCDetailsRussia):
|
||
return profile.country_details.inn
|
||
return None
|
||
except KYCProfile.DoesNotExist:
|
||
return None
|
||
|
||
|
||
def aggregate_company_data(documents: list) -> dict:
|
||
"""Aggregate data from multiple source documents into summary."""
|
||
if not documents:
|
||
return {}
|
||
|
||
summary = {
|
||
"inn": None,
|
||
"ogrn": None,
|
||
"name": None,
|
||
"company_type": None,
|
||
"registration_year": None,
|
||
"is_active": True,
|
||
"address": None,
|
||
"director": None,
|
||
"capital": None,
|
||
"activities": [],
|
||
"sources": [],
|
||
"last_updated": None,
|
||
}
|
||
|
||
for doc in documents:
|
||
source = doc.get("source", "unknown")
|
||
summary["sources"].append(source)
|
||
|
||
data = doc.get("data", {})
|
||
|
||
# Extract common fields
|
||
if not summary["inn"]:
|
||
summary["inn"] = doc.get("inn")
|
||
|
||
if not summary["ogrn"] and data.get("ogrn"):
|
||
summary["ogrn"] = data["ogrn"]
|
||
|
||
if not summary["name"] and data.get("name"):
|
||
summary["name"] = data["name"]
|
||
|
||
# Parse company type from name
|
||
if not summary["company_type"] and summary["name"]:
|
||
name = summary["name"].upper()
|
||
if "ООО" in name or "ОБЩЕСТВО С ОГРАНИЧЕННОЙ" in name:
|
||
summary["company_type"] = "ООО"
|
||
elif "АО" in name or "АКЦИОНЕРНОЕ ОБЩЕСТВО" in name:
|
||
summary["company_type"] = "АО"
|
||
elif "ИП" in name or "ИНДИВИДУАЛЬНЫЙ ПРЕДПРИНИМАТЕЛЬ" in name:
|
||
summary["company_type"] = "ИП"
|
||
elif "ПАО" in name:
|
||
summary["company_type"] = "ПАО"
|
||
|
||
# Track last update
|
||
collected_at = doc.get("collected_at")
|
||
if collected_at:
|
||
if not summary["last_updated"] or collected_at > summary["last_updated"]:
|
||
summary["last_updated"] = collected_at
|
||
|
||
return summary
|
||
|
||
|
||
class CompanyTeaserType(graphene.ObjectType):
|
||
"""Public company data (teaser)."""
|
||
|
||
company_type = graphene.String(description="Company type: ООО, АО, ИП, etc.")
|
||
registration_year = graphene.Int(description="Year of registration")
|
||
is_active = graphene.Boolean(description="Is company active")
|
||
sources_count = graphene.Int(description="Number of data sources")
|
||
|
||
|
||
class CompanyFullType(graphene.ObjectType):
|
||
"""Full company data (requires auth)."""
|
||
|
||
inn = graphene.String()
|
||
ogrn = graphene.String()
|
||
name = graphene.String()
|
||
company_type = graphene.String()
|
||
registration_year = graphene.Int()
|
||
is_active = graphene.Boolean()
|
||
address = graphene.String()
|
||
director = graphene.String()
|
||
capital = graphene.String()
|
||
activities = graphene.List(graphene.String)
|
||
sources = graphene.List(graphene.String)
|
||
last_updated = graphene.DateTime()
|
||
|
||
|
||
class PublicQuery(graphene.ObjectType):
|
||
"""Public queries - no authentication required."""
|
||
|
||
# Query by KYC Profile UUID (preferred - used by frontend)
|
||
company_teaser_by_profile = graphene.Field(
|
||
CompanyTeaserType,
|
||
profile_uuid=graphene.String(required=True),
|
||
description="Get public company teaser data by KYC Profile UUID",
|
||
)
|
||
|
||
company_full_by_profile = graphene.Field(
|
||
CompanyFullType,
|
||
profile_uuid=graphene.String(required=True),
|
||
description="Get full company data by KYC Profile UUID (requires auth)",
|
||
)
|
||
|
||
# Query by INN (legacy/direct)
|
||
company_teaser = graphene.Field(
|
||
CompanyTeaserType,
|
||
inn=graphene.String(required=True),
|
||
description="Get public company teaser data by INN",
|
||
)
|
||
|
||
company_full = graphene.Field(
|
||
CompanyFullType,
|
||
inn=graphene.String(required=True),
|
||
description="Get full company data by INN (requires auth)",
|
||
)
|
||
|
||
health = graphene.String()
|
||
|
||
def resolve_health(self, info):
|
||
return "ok"
|
||
|
||
def resolve_company_teaser_by_profile(self, info, profile_uuid: str):
|
||
"""Return public teaser data by KYC Profile UUID."""
|
||
inn = get_inn_by_profile_uuid(profile_uuid)
|
||
if not inn:
|
||
return None
|
||
|
||
documents = get_company_documents(inn)
|
||
if not documents:
|
||
return None
|
||
|
||
summary = aggregate_company_data(documents)
|
||
|
||
return CompanyTeaserType(
|
||
company_type=summary.get("company_type"),
|
||
registration_year=summary.get("registration_year"),
|
||
is_active=summary.get("is_active", True),
|
||
sources_count=len(summary.get("sources", [])),
|
||
)
|
||
|
||
def resolve_company_full_by_profile(self, info, profile_uuid: str):
|
||
"""Return full company data by KYC Profile UUID (requires auth)."""
|
||
# Check authentication
|
||
user_id = getattr(info.context, 'user_id', None)
|
||
if not user_id:
|
||
return None # Not authenticated
|
||
|
||
inn = get_inn_by_profile_uuid(profile_uuid)
|
||
if not inn:
|
||
return None
|
||
|
||
documents = get_company_documents(inn)
|
||
if not documents:
|
||
return None
|
||
|
||
summary = aggregate_company_data(documents)
|
||
|
||
return CompanyFullType(
|
||
inn=summary.get("inn"),
|
||
ogrn=summary.get("ogrn"),
|
||
name=summary.get("name"),
|
||
company_type=summary.get("company_type"),
|
||
registration_year=summary.get("registration_year"),
|
||
is_active=summary.get("is_active", True),
|
||
address=summary.get("address"),
|
||
director=summary.get("director"),
|
||
capital=summary.get("capital"),
|
||
activities=summary.get("activities", []),
|
||
sources=summary.get("sources", []),
|
||
last_updated=summary.get("last_updated"),
|
||
)
|
||
|
||
def resolve_company_teaser(self, info, inn: str):
|
||
"""Return public teaser data by INN (no auth required)."""
|
||
documents = get_company_documents(inn)
|
||
if not documents:
|
||
return None
|
||
|
||
summary = aggregate_company_data(documents)
|
||
|
||
return CompanyTeaserType(
|
||
company_type=summary.get("company_type"),
|
||
registration_year=summary.get("registration_year"),
|
||
is_active=summary.get("is_active", True),
|
||
sources_count=len(summary.get("sources", [])),
|
||
)
|
||
|
||
def resolve_company_full(self, info, inn: str):
|
||
"""Return full company data by INN (requires auth)."""
|
||
# Check authentication
|
||
user_id = getattr(info.context, 'user_id', None)
|
||
if not user_id:
|
||
return None # Not authenticated
|
||
|
||
documents = get_company_documents(inn)
|
||
if not documents:
|
||
return None
|
||
|
||
summary = aggregate_company_data(documents)
|
||
|
||
return CompanyFullType(
|
||
inn=summary.get("inn"),
|
||
ogrn=summary.get("ogrn"),
|
||
name=summary.get("name"),
|
||
company_type=summary.get("company_type"),
|
||
registration_year=summary.get("registration_year"),
|
||
is_active=summary.get("is_active", True),
|
||
address=summary.get("address"),
|
||
director=summary.get("director"),
|
||
capital=summary.get("capital"),
|
||
activities=summary.get("activities", []),
|
||
sources=summary.get("sources", []),
|
||
last_updated=summary.get("last_updated"),
|
||
)
|
||
|
||
|
||
public_schema = graphene.Schema(query=PublicQuery)
|