Add service platform foundation
This commit is contained in:
48
app/services/ocr_provider.py
Normal file
48
app/services/ocr_provider.py
Normal file
@@ -0,0 +1,48 @@
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from app.services.vehicle_identity import normalize_license_plate, validate_vin
|
||||
|
||||
|
||||
@dataclass
|
||||
class OcrCandidate:
|
||||
type: str
|
||||
value: str
|
||||
confidence: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class OcrResult:
|
||||
recognized_text: str
|
||||
candidates: list[OcrCandidate]
|
||||
|
||||
|
||||
class OCRProvider:
|
||||
async def recognize(self, content: bytes, filename: str | None = None) -> OcrResult:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class StubOCRProvider(OCRProvider):
|
||||
async def recognize(self, content: bytes, filename: str | None = None) -> OcrResult:
|
||||
text = " ".join(
|
||||
[
|
||||
filename or "",
|
||||
content.decode("utf-8", errors="ignore"),
|
||||
]
|
||||
)
|
||||
compact = re.sub(r"\s+", " ", text).strip()
|
||||
candidates: list[OcrCandidate] = []
|
||||
for raw in re.findall(r"\b[A-HJ-NPR-Z0-9]{17}\b", compact.upper()):
|
||||
try:
|
||||
candidates.append(OcrCandidate(type="vin", value=validate_vin(raw) or raw, confidence=0.84))
|
||||
except ValueError:
|
||||
continue
|
||||
for raw in re.findall(r"\b[0-9A-ZА-Я가-힣][0-9A-ZА-Я가-힣\-\s]{4,10}\b", compact.upper()):
|
||||
normalized = normalize_license_plate(raw)
|
||||
if normalized and 5 <= len(normalized) <= 10 and not any(item.value == normalized for item in candidates):
|
||||
candidates.append(OcrCandidate(type="license_plate", value=normalized, confidence=0.62))
|
||||
return OcrResult(recognized_text=compact, candidates=candidates[:8])
|
||||
|
||||
|
||||
def get_ocr_provider() -> OCRProvider:
|
||||
return StubOCRProvider()
|
||||
44
app/services/vehicle_identity.py
Normal file
44
app/services/vehicle_identity.py
Normal file
@@ -0,0 +1,44 @@
|
||||
import re
|
||||
|
||||
VIN_RE = re.compile(r"^[A-HJ-NPR-Z0-9]{17}$")
|
||||
|
||||
|
||||
def normalize_vin(value: str | None) -> str | None:
|
||||
if not value:
|
||||
return None
|
||||
normalized = re.sub(r"[\s-]+", "", value).upper()
|
||||
return normalized or None
|
||||
|
||||
|
||||
def validate_vin(value: str | None) -> str | None:
|
||||
normalized = normalize_vin(value)
|
||||
if normalized is None:
|
||||
return None
|
||||
if not VIN_RE.match(normalized):
|
||||
raise ValueError("VIN must contain 17 characters and cannot include I, O, or Q")
|
||||
return normalized
|
||||
|
||||
|
||||
def normalize_license_plate(value: str | None) -> str | None:
|
||||
if not value:
|
||||
return None
|
||||
normalized = re.sub(r"[\s\-_.]+", "", value).upper()
|
||||
return normalized or None
|
||||
|
||||
|
||||
def mask_vin(value: str | None) -> str | None:
|
||||
normalized = normalize_vin(value)
|
||||
if not normalized:
|
||||
return None
|
||||
if len(normalized) <= 6:
|
||||
return "*" * len(normalized)
|
||||
return f"{normalized[:3]}{'*' * 10}{normalized[-4:]}"
|
||||
|
||||
|
||||
def mask_license_plate(value: str | None) -> str | None:
|
||||
normalized = normalize_license_plate(value)
|
||||
if not normalized:
|
||||
return None
|
||||
if len(normalized) <= 3:
|
||||
return "*" * len(normalized)
|
||||
return f"{normalized[:2]}{'*' * max(len(normalized) - 4, 2)}{normalized[-2:]}"
|
||||
Reference in New Issue
Block a user