Spaces:
Sleeping
Sleeping
File size: 2,044 Bytes
fa20fa2 02109ec fa20fa2 02109ec fa20fa2 02109ec cc2e026 bfe7c08 d57b994 fa20fa2 02109ec fa20fa2 02109ec fa20fa2 02109ec fa20fa2 02109ec fa20fa2 bfe7c08 cc2e026 fa20fa2 cc2e026 bfe7c08 02109ec |
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 |
import os
from typing import Annotated
from fastapi import FastAPI, File, UploadFile, Form
from ttp import ttp
from app.core.cognito import Cognito
from app.core.s3 import S3
from app.core.supabase import supabase_client
from app.core.template import Template
app = FastAPI()
@app.post("/data")
def data(
bucket: str,
project: str,
local_date: str,
env_file: UploadFile = File(...)
):
server_folder = f"{project}/{local_date}"
local_folder = f"data/{server_folder}"
s3 = S3.client(file=env_file)
response = s3.list_objects_v2(Bucket=bucket, Prefix=server_folder)
if not os.path.exists(local_folder):
os.makedirs(local_folder)
with open(f"{local_folder}/data.txt", "ab") as f:
for obj in response.get('Contents', []):
s3.download_fileobj(bucket, obj['Key'], f)
return {"message": "success"}
@app.get("/data")
def data(
project: str,
local_date: str,
process: str
):
local_folder = f"data/{project}/{local_date}"
template = Template.read(process)
values = {"project": project, "local_date": local_date, "process": process}
template = template.format_map(values)
parser = ttp(data=f"{local_folder}/data.txt", template=template)
parser.parse()
result = parser.result()[0][0]
for i in range(0, len(result), 7777):
try:
lote = result[i:i + 7777]
response = supabase_client().table("log_back_office").upsert(lote, ignore_duplicates=True).execute()
print(f"subiendo {len(response.data)}/{len(result)} desde {i}")
pass
except Exception as e:
print(f"error desde {i}: {e}")
return result
@app.post("/auth")
def auth(
username: Annotated[str, Form()],
password: Annotated[str, Form()],
env_file: UploadFile = File(...)
):
cognito = Cognito.client(file=env_file)
response = Cognito.auth(
client=cognito,
username=username,
password=password
)
return response
|