You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
469 lines
15 KiB
Python
469 lines
15 KiB
Python
import hashlib
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import func
|
|
|
|
from app.db.mock_database import SessionMockLocal
|
|
from app.db.mock_models import ReviewSchedule
|
|
from app.services.domain.common import parse_bool
|
|
from app.services.domain.tool_errors import build_tool_error, raise_tool_http_error
|
|
|
|
# Responsabilidade: tudo que é regra de revisão.
|
|
|
|
|
|
def _base_review_price_by_model(modelo: str) -> float:
|
|
text = (modelo or "").lower()
|
|
premium_brands = ("bmw", "audi", "mercedes", "volvo", "land rover", "lexus")
|
|
if any(brand in text for brand in premium_brands):
|
|
return 1200.0
|
|
if any(tag in text for tag in ("suv", "pickup", "caminhonete")):
|
|
return 900.0
|
|
if "sedan" in text:
|
|
return 750.0
|
|
if "hatch" in text:
|
|
return 650.0
|
|
return 700.0
|
|
|
|
|
|
def _calculate_review_price(
|
|
modelo: str,
|
|
ano: int,
|
|
km: int,
|
|
revisao_previa_concessionaria: bool,
|
|
) -> float:
|
|
base = _base_review_price_by_model(modelo)
|
|
ano_atual = datetime.now().year
|
|
idade = max(0, ano_atual - int(ano))
|
|
fator_idade = 1.0 + min(idade * 0.02, 0.30)
|
|
|
|
km_int = max(0, int(km))
|
|
if km_int <= 20000:
|
|
adicional_km = 0.0
|
|
elif km_int <= 60000:
|
|
adicional_km = 150.0
|
|
elif km_int <= 100000:
|
|
adicional_km = 300.0
|
|
else:
|
|
adicional_km = 500.0
|
|
|
|
subtotal = (base * fator_idade) + adicional_km
|
|
if revisao_previa_concessionaria:
|
|
subtotal *= 0.90
|
|
return round(max(subtotal, 300.0), 2)
|
|
|
|
|
|
def _parse_tzinfo(offset: str | None) -> timezone | None:
|
|
if not offset:
|
|
return None
|
|
if offset == "Z":
|
|
return timezone.utc
|
|
sign = 1 if offset[0] == "+" else -1
|
|
hours = int(offset[1:3])
|
|
minutes = int(offset[4:6])
|
|
return timezone(sign * timedelta(hours=hours, minutes=minutes))
|
|
|
|
|
|
def parse_review_datetime(value: str) -> datetime:
|
|
text = (value or "").strip()
|
|
if not text:
|
|
raise ValueError("data_hora vazia")
|
|
|
|
normalized = re.sub(r"\s+[aàáâã]s\s+", " ", text, flags=re.IGNORECASE)
|
|
for candidate in (text, normalized):
|
|
try:
|
|
return datetime.fromisoformat(candidate.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
pass
|
|
|
|
patterns = (
|
|
r"^(?P<day>\d{1,2})[/-](?P<month>\d{1,2})[/-](?P<year>\d{4})\s+"
|
|
r"(?P<hour>\d{1,2}):(?P<minute>\d{2})(?::(?P<second>\d{2}))?"
|
|
r"(?:\s*(?P<tz>Z|[+-]\d{2}:\d{2}))?$",
|
|
r"^(?P<year>\d{4})[/-](?P<month>\d{1,2})[/-](?P<day>\d{1,2})\s+"
|
|
r"(?P<hour>\d{1,2}):(?P<minute>\d{2})(?::(?P<second>\d{2}))?"
|
|
r"(?:\s*(?P<tz>Z|[+-]\d{2}:\d{2}))?$",
|
|
)
|
|
|
|
for pattern in patterns:
|
|
match = re.match(pattern, normalized)
|
|
if not match:
|
|
continue
|
|
parts = match.groupdict()
|
|
return datetime(
|
|
year=int(parts["year"]),
|
|
month=int(parts["month"]),
|
|
day=int(parts["day"]),
|
|
hour=int(parts["hour"]),
|
|
minute=int(parts["minute"]),
|
|
second=int(parts["second"] or 0),
|
|
tzinfo=_parse_tzinfo(parts.get("tz")),
|
|
)
|
|
|
|
raise ValueError("formato invalido")
|
|
|
|
|
|
def _normalize_review_slot(value: datetime) -> datetime:
|
|
return value.replace(second=0, microsecond=0)
|
|
|
|
|
|
def _format_datetime_pt_br(value: datetime) -> str:
|
|
return value.strftime("%d/%m/%Y as %H:%M")
|
|
|
|
|
|
def _find_next_available_review_slot(
|
|
db,
|
|
requested_dt: datetime,
|
|
max_attempts: int = 16,
|
|
step_minutes: int = 30,
|
|
) -> datetime | None:
|
|
for attempt in range(1, max_attempts + 1):
|
|
candidate = requested_dt + timedelta(minutes=step_minutes * attempt)
|
|
ocupado = (
|
|
db.query(ReviewSchedule)
|
|
.filter(ReviewSchedule.data_hora == candidate)
|
|
.filter(func.lower(ReviewSchedule.status) != "cancelado")
|
|
.first()
|
|
)
|
|
if not ocupado:
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def build_review_conflict_detail(
|
|
requested_dt: datetime,
|
|
suggested_dt: datetime | None = None,
|
|
) -> dict[str, Any]:
|
|
message = (
|
|
f"O horario {_format_datetime_pt_br(requested_dt)} ja esta ocupado. "
|
|
f"Posso agendar em {_format_datetime_pt_br(suggested_dt)}."
|
|
if suggested_dt is not None
|
|
else (
|
|
f"O horario {_format_datetime_pt_br(requested_dt)} ja esta ocupado e nao encontrei "
|
|
"disponibilidade nas proximas 8 horas."
|
|
)
|
|
)
|
|
return build_tool_error(
|
|
code="review_schedule_conflict",
|
|
message=message,
|
|
retryable=True,
|
|
field="data_hora",
|
|
meta={
|
|
"requested_iso": requested_dt.isoformat(),
|
|
"suggested_iso": suggested_dt.isoformat() if suggested_dt is not None else None,
|
|
},
|
|
)
|
|
|
|
|
|
async def agendar_revisao(
|
|
placa: str,
|
|
data_hora: str,
|
|
modelo: str,
|
|
ano: int,
|
|
km: int,
|
|
revisao_previa_concessionaria: bool,
|
|
user_id: int | None = None,
|
|
) -> dict[str, Any]:
|
|
try:
|
|
ano_int = int(ano)
|
|
km_int = int(km)
|
|
except (TypeError, ValueError):
|
|
raise_tool_http_error(
|
|
status_code=400,
|
|
code="invalid_vehicle_data",
|
|
message="ano e km devem ser valores inteiros validos.",
|
|
retryable=True,
|
|
)
|
|
|
|
ano_atual = datetime.now().year
|
|
if ano_int < 1980 or ano_int > ano_atual + 1:
|
|
raise_tool_http_error(
|
|
status_code=400,
|
|
code="invalid_year",
|
|
message=f"ano invalido. Informe entre 1980 e {ano_atual + 1}.",
|
|
retryable=True,
|
|
field="ano",
|
|
)
|
|
if km_int < 0:
|
|
raise_tool_http_error(
|
|
status_code=400,
|
|
code="invalid_km",
|
|
message="km invalido. Informe um valor maior ou igual a zero.",
|
|
retryable=True,
|
|
field="km",
|
|
)
|
|
|
|
try:
|
|
dt = _normalize_review_slot(parse_review_datetime(data_hora))
|
|
except ValueError:
|
|
raise_tool_http_error(
|
|
status_code=400,
|
|
code="invalid_review_datetime",
|
|
message=(
|
|
"data_hora invalida. Exemplos aceitos: "
|
|
"2026-03-10T09:00:00-03:00, 2026-03-10 09:00, 10/03/2026 09:00, "
|
|
"10/03/2026 as 09:00."
|
|
),
|
|
retryable=True,
|
|
field="data_hora",
|
|
)
|
|
|
|
placa_normalizada = placa.upper()
|
|
revisao_previa = parse_bool(revisao_previa_concessionaria)
|
|
valor_revisao = _calculate_review_price(
|
|
modelo=modelo,
|
|
ano=ano_int,
|
|
km=km_int,
|
|
revisao_previa_concessionaria=revisao_previa,
|
|
)
|
|
dt_canonical = dt.isoformat()
|
|
entropy = hashlib.md5(f"{user_id}:{placa_normalizada}:{dt_canonical}".encode("utf-8")).hexdigest()[:8].upper()
|
|
protocolo = f"REV-{dt.strftime('%Y%m%d')}-{entropy}"
|
|
|
|
db = SessionMockLocal()
|
|
try:
|
|
conflito_horario = (
|
|
db.query(ReviewSchedule)
|
|
.filter(ReviewSchedule.data_hora == dt)
|
|
.filter(func.lower(ReviewSchedule.status) != "cancelado")
|
|
.first()
|
|
)
|
|
if conflito_horario:
|
|
proximo_horario = _find_next_available_review_slot(db=db, requested_dt=dt)
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=build_review_conflict_detail(
|
|
requested_dt=dt,
|
|
suggested_dt=proximo_horario,
|
|
),
|
|
)
|
|
|
|
existente = db.query(ReviewSchedule).filter(ReviewSchedule.protocolo == protocolo).first()
|
|
if existente:
|
|
return {
|
|
"protocolo": existente.protocolo,
|
|
"user_id": existente.user_id,
|
|
"placa": existente.placa,
|
|
"data_hora": existente.data_hora.isoformat(),
|
|
"status": existente.status,
|
|
"modelo": modelo,
|
|
"ano": ano_int,
|
|
"km": km_int,
|
|
"revisao_previa_concessionaria": revisao_previa,
|
|
"valor_revisao": valor_revisao,
|
|
}
|
|
|
|
agendamento = ReviewSchedule(
|
|
protocolo=protocolo,
|
|
user_id=user_id,
|
|
placa=placa_normalizada,
|
|
data_hora=dt,
|
|
status="agendado",
|
|
)
|
|
db.add(agendamento)
|
|
db.commit()
|
|
db.refresh(agendamento)
|
|
|
|
return {
|
|
"protocolo": agendamento.protocolo,
|
|
"user_id": agendamento.user_id,
|
|
"placa": agendamento.placa,
|
|
"data_hora": agendamento.data_hora.isoformat(),
|
|
"status": agendamento.status,
|
|
"modelo": modelo,
|
|
"ano": ano_int,
|
|
"km": km_int,
|
|
"revisao_previa_concessionaria": revisao_previa,
|
|
"valor_revisao": valor_revisao,
|
|
}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
async def listar_agendamentos_revisao(
|
|
user_id: int | None = None,
|
|
placa: str | None = None,
|
|
status: str | None = None,
|
|
limite: int | None = 20,
|
|
) -> list[dict[str, Any]]:
|
|
if user_id is None:
|
|
raise_tool_http_error(
|
|
status_code=400,
|
|
code="missing_user_id",
|
|
message="Informe user_id para listar seus agendamentos de revisao.",
|
|
retryable=False,
|
|
)
|
|
|
|
placa_normalizada = placa.upper().strip() if placa else None
|
|
status_normalizado = status.lower().strip() if status else None
|
|
|
|
try:
|
|
limite_int = int(limite) if limite is not None else 20
|
|
except (TypeError, ValueError):
|
|
limite_int = 20
|
|
limite_int = max(1, min(limite_int, 100))
|
|
|
|
db = SessionMockLocal()
|
|
try:
|
|
query = db.query(ReviewSchedule).filter(ReviewSchedule.user_id == user_id)
|
|
if placa_normalizada:
|
|
query = query.filter(ReviewSchedule.placa == placa_normalizada)
|
|
if status_normalizado:
|
|
query = query.filter(func.lower(ReviewSchedule.status) == status_normalizado)
|
|
|
|
agendamentos = query.order_by(ReviewSchedule.data_hora.asc()).limit(limite_int).all()
|
|
return [
|
|
{
|
|
"protocolo": row.protocolo,
|
|
"user_id": row.user_id,
|
|
"placa": row.placa,
|
|
"data_hora": row.data_hora.isoformat(),
|
|
"status": row.status,
|
|
"created_at": row.created_at.isoformat() if row.created_at else None,
|
|
}
|
|
for row in agendamentos
|
|
]
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
async def cancelar_agendamento_revisao(
|
|
protocolo: str,
|
|
motivo: str | None = None,
|
|
user_id: int | None = None,
|
|
) -> dict[str, Any]:
|
|
if user_id is None:
|
|
raise_tool_http_error(
|
|
status_code=400,
|
|
code="missing_user_id",
|
|
message="Informe user_id para cancelar seu agendamento de revisao.",
|
|
retryable=False,
|
|
)
|
|
|
|
db = SessionMockLocal()
|
|
try:
|
|
agendamento = (
|
|
db.query(ReviewSchedule)
|
|
.filter(ReviewSchedule.protocolo == protocolo)
|
|
.filter(ReviewSchedule.user_id == user_id)
|
|
.first()
|
|
)
|
|
if not agendamento:
|
|
raise_tool_http_error(
|
|
status_code=404,
|
|
code="review_not_found",
|
|
message="Agendamento de revisao nao encontrado para este usuario.",
|
|
retryable=True,
|
|
field="protocolo",
|
|
)
|
|
|
|
if agendamento.status.lower() == "cancelado":
|
|
return {
|
|
"protocolo": agendamento.protocolo,
|
|
"user_id": agendamento.user_id,
|
|
"placa": agendamento.placa,
|
|
"data_hora": agendamento.data_hora.isoformat(),
|
|
"status": agendamento.status,
|
|
"motivo": motivo,
|
|
}
|
|
|
|
agendamento.status = "cancelado"
|
|
db.commit()
|
|
db.refresh(agendamento)
|
|
return {
|
|
"protocolo": agendamento.protocolo,
|
|
"user_id": agendamento.user_id,
|
|
"placa": agendamento.placa,
|
|
"data_hora": agendamento.data_hora.isoformat(),
|
|
"status": agendamento.status,
|
|
"motivo": motivo,
|
|
}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
async def editar_data_revisao(
|
|
protocolo: str,
|
|
nova_data_hora: str,
|
|
user_id: int | None = None,
|
|
) -> dict[str, Any]:
|
|
if user_id is None:
|
|
raise_tool_http_error(
|
|
status_code=400,
|
|
code="missing_user_id",
|
|
message="Informe user_id para editar seu agendamento de revisao.",
|
|
retryable=False,
|
|
)
|
|
|
|
try:
|
|
nova_data = _normalize_review_slot(parse_review_datetime(nova_data_hora))
|
|
except ValueError:
|
|
raise_tool_http_error(
|
|
status_code=400,
|
|
code="invalid_review_datetime",
|
|
message=(
|
|
"nova_data_hora invalida. Exemplos aceitos: "
|
|
"2026-03-10T09:00:00-03:00, 2026-03-10 09:00, 10/03/2026 09:00, "
|
|
"10/03/2026 as 09:00."
|
|
),
|
|
retryable=True,
|
|
field="nova_data_hora",
|
|
)
|
|
|
|
db = SessionMockLocal()
|
|
try:
|
|
agendamento = (
|
|
db.query(ReviewSchedule)
|
|
.filter(ReviewSchedule.protocolo == protocolo)
|
|
.filter(ReviewSchedule.user_id == user_id)
|
|
.first()
|
|
)
|
|
if not agendamento:
|
|
raise_tool_http_error(
|
|
status_code=404,
|
|
code="review_not_found",
|
|
message="Agendamento de revisao nao encontrado para este usuario.",
|
|
retryable=True,
|
|
field="protocolo",
|
|
)
|
|
|
|
if agendamento.status.lower() == "cancelado":
|
|
raise_tool_http_error(
|
|
status_code=400,
|
|
code="review_already_cancelled",
|
|
message="Nao e possivel editar um agendamento ja cancelado.",
|
|
retryable=False,
|
|
)
|
|
|
|
conflito = (
|
|
db.query(ReviewSchedule)
|
|
.filter(ReviewSchedule.id != agendamento.id)
|
|
.filter(ReviewSchedule.data_hora == nova_data)
|
|
.filter(func.lower(ReviewSchedule.status) != "cancelado")
|
|
.first()
|
|
)
|
|
if conflito:
|
|
proximo_horario = _find_next_available_review_slot(db=db, requested_dt=nova_data)
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=build_review_conflict_detail(
|
|
requested_dt=nova_data,
|
|
suggested_dt=proximo_horario,
|
|
),
|
|
)
|
|
|
|
agendamento.data_hora = nova_data
|
|
db.commit()
|
|
db.refresh(agendamento)
|
|
return {
|
|
"protocolo": agendamento.protocolo,
|
|
"user_id": agendamento.user_id,
|
|
"placa": agendamento.placa,
|
|
"data_hora": agendamento.data_hora.isoformat(),
|
|
"status": agendamento.status,
|
|
}
|
|
finally:
|
|
db.close()
|