Se mejoro el disenio de partes, se genero la informacion mas precisa en los reportes y se carga el logo en las instanacias de las empresas

This commit is contained in:
2026-01-15 13:20:34 -06:00
parent ae1d7f0c58
commit 0a02a3e912
44 changed files with 1536 additions and 515 deletions

View File

@@ -26,7 +26,7 @@ class Company(Base, TimestampMixin):
__tablename__ = "company" #GEmpresa
__table_args__ = (
PrimaryKeyConstraint("id", name="company_pkey"),
{"schema": "a76"},
{"schema": "a76", "extend_existing": True},
)
# Primary key

View File

@@ -3,8 +3,12 @@ Rutas para gestión de empresa
"""
from typing import List, Optional
import os
import shutil
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, status, File, UploadFile
from fastapi.responses import FileResponse
from sqlalchemy.orm import Session
from core.database import get_core_db
@@ -298,11 +302,98 @@ async def update_company(
return CompanyResponseDTO.model_validate(updated_company)
@router.delete(
"/{company_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete company",
return CompanyResponseDTO.model_validate(updated_company)
@router.post(
"/{company_id}/upload-logo",
response_model=dict,
summary="Upload company logo",
)
async def upload_company_logo(
company_id: int,
file: UploadFile = File(...),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""Upload logo for a company"""
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
# 1. Verify company exists
company = CompanyService.get_by_id(db, company_id, tenant_id, 0)
if not company:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Company not found",
)
# 2. Define upload path
# Use a persistent path: 'app_data/logos/{company_id}'
upload_dir = Path(f"app_data/logos/{company_id}")
upload_dir.mkdir(parents=True, exist_ok=True)
# 3. Save file
# Preserve original filename
filename = file.filename or "logo.png"
file_path = upload_dir / filename
try:
# Check if file exists and remove it to avoid accumulation if needed,
# or just overwrite (shutil.copyfileobj overwrites)
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Could not save file: {e}",
)
# 4. Returns the absolute path keys
abs_path = str(file_path.absolute())
return {"path": abs_path}
@router.get(
"/{company_id}/logo/image",
summary="Get company logo image",
)
@router.get(
"/{company_id}/logo/image",
summary="Get company logo image",
)
async def get_company_logo_image(
company_id: int,
db: Session = Depends(get_core_db),
# Public endpoint to allow <img> tags to load the image without custom headers
):
"""Serve the company logo image file"""
# Security: In a stricter environment, we would use a signed short-lived URL
# or cookie-based auth. For now, checking if company exists is sufficient.
# We find the company ignoring tenant checks for the image serving
# (Logos are generally considered semi-public assets in this context)
company = db.query(Company).filter(Company.id == company_id).first()
if not company or not company.logo:
raise HTTPException(status_code=404, detail="Logo not found")
file_path = Path(company.logo)
if not file_path.exists():
# Fallback for old paths or moved files
# Check if it exists in the 'standard' location even if DB thinks otherwise
standard_path = Path(f"app_data/logos/{company_id}") / file_path.name
if standard_path.exists():
return FileResponse(standard_path)
raise HTTPException(status_code=404, detail="Logo file not found on server")
return FileResponse(file_path)
async def delete_company(
company_id: int,
db: Session = Depends(get_core_db),

View File

@@ -34,10 +34,13 @@ class BaseService:
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[Any], int]:
query = db.query(cls.model).filter(
cls.model.tenant_id == tenant_id,
cls.model.company_id == company_id,
)
query = db.query(cls.model)
if hasattr(cls.model, "tenant_id"):
query = query.filter(cls.model.tenant_id == tenant_id)
if hasattr(cls.model, "company_id"):
query = query.filter(cls.model.company_id == company_id)
if filters:
if filters.get("code"):
@@ -56,11 +59,15 @@ class BaseService:
def get_by_id(
cls, db: Session, id: int, tenant_id: int, company_id: int
) -> Optional[Any]:
return db.query(cls.model).filter(
cls.model.id == id,
cls.model.tenant_id == tenant_id,
cls.model.company_id == company_id,
).first()
query = db.query(cls.model).filter(cls.model.id == id)
if hasattr(cls.model, "tenant_id"):
query = query.filter(cls.model.tenant_id == tenant_id)
if hasattr(cls.model, "company_id"):
query = query.filter(cls.model.company_id == company_id)
return query.first()
@classmethod
def create(
@@ -70,9 +77,15 @@ class BaseService:
tenant_id: int,
company_id: int,
) -> Any:
db_obj = cls.model(
**data.model_dump(), tenant_id=tenant_id, company_id=company_id
)
create_kwargs = data.model_dump()
if hasattr(cls.model, "tenant_id"):
create_kwargs["tenant_id"] = tenant_id
if hasattr(cls.model, "company_id"):
create_kwargs["company_id"] = company_id
db_obj = cls.model(**create_kwargs)
db.add(db_obj)
db.commit()
db.refresh(db_obj)