Spaces:
Running
on
Zero
Running
on
Zero
File size: 3,033 Bytes
e368cec 944dd2b e368cec 350d553 e368cec 8dcd2f5 e368cec 835bec4 e368cec 835bec4 e368cec 350d553 944dd2b 350d553 |
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 |
from fastapi import FastAPI, File, UploadFile, Form, APIRouter
from typing import Optional
import json
import os
import aiofiles
from .log_utils import build_logger
from .constants import LOG_SERVER_SUBDOMAIN, APPEND_JSON, SAVE_IMAGE, SAVE_VIDEO, SAVE_LOG
logger = build_logger("log_server", "log_server.log", add_remote_handler=False)
app = APIRouter(prefix=LOG_SERVER_SUBDOMAIN)
@app.post(f"/{APPEND_JSON}")
async def append_json(json_str: str = Form(...), file_name: str = Form(...)):
"""
Appends a JSON string to a specified file.
"""
# Convert the string back to a JSON object (dict)
data = json.loads(json_str)
# Append the data to the specified file
if os.path.dirname(file_name):
os.makedirs(os.path.dirname(file_name), exist_ok=True)
async with aiofiles.open(file_name, mode='a') as f:
await f.write(json.dumps(data) + "\n")
logger.info(f"Appended 1 JSON object to {file_name}")
return {"message": "JSON data appended successfully"}
@app.post(f"/{SAVE_IMAGE}")
async def save_image(image: UploadFile = File(...), image_path: str = Form(...)):
"""
Saves an uploaded image to the specified path.
"""
# Note: 'image_path' should include the file name and extension for the image to be saved.
if os.path.dirname(image_path):
os.makedirs(os.path.dirname(image_path), exist_ok=True)
async with aiofiles.open(image_path, mode='wb') as f:
content = await image.read() # Read the content of the uploaded image
await f.write(content) # Write the image content to a file
logger.info(f"Image saved successfully at {image_path}")
return {"message": f"Image saved successfully at {image_path}"}
@app.post(f"/{SAVE_VIDEO}")
async def save_video(video: UploadFile = File(...), video_path: str = Form(...)):
"""
Saves an uploaded video to the specified path.
"""
# Note: 'video_path' should include the file name and extension for the video to be saved.
if os.path.dirname(video_path):
os.makedirs(os.path.dirname(video_path), exist_ok=True)
async with aiofiles.open(video_path, mode='wb') as f:
content = await video.read() # Read the content of the uploaded video
await f.write(content) # Write the video content to a file
logger.info(f"Video saved successfully at {video_path}")
return {"message": f"Image saved successfully at {video_path}"}
@app.post(f"/{SAVE_LOG}")
async def save_log(message: str = Form(...), log_path: str = Form(...)):
"""
Save a log message to a specified log file on the server.
"""
# Ensure the directory for the log file exists
if os.path.dirname(log_path):
os.makedirs(os.path.dirname(log_path), exist_ok=True)
# Append the log message to the specified log file
async with aiofiles.open(log_path, mode='a') as f:
await f.write(f"{message}\n")
logger.info(f"Romote log message saved to {log_path}")
return {"message": f"Log message saved successfully to {log_path}"} |