# main.py import shutil import uvicorn from fastapi import FastAPI, UploadFile, File, HTTPException, Query, Request from fastapi.responses import PlainTextResponse, JSONResponse, FileResponse, Response from fastapi.middleware.cors import CORSMiddleware from urllib.parse import urlencode import time import os import stat import fnmatch import re from typing import Iterator, List from glob import glob from datetime import datetime #from ums_db import * import ls_html #import pwa # --- Setup --- # Create a FastAPI app instance app = FastAPI( title="zshlo", description="zshlo server", version="1.1.0" ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], # Allows all origins, you can specify a list of allowed origins allow_credentials=True, allow_methods=["*"], # Allows all methods (GET, POST, etc.) allow_headers=["*"], # Allows all headers ) # Define the directory where uploaded files will be stored ZFS_DIR = '../ZFS/' #Path("./") #'./' log_file = './v2FS/log.txt' # Ensure the upload directory exists #ZFS_DIR.mkdir(parents=True, exist_ok=True) fileTypes = { #img file '.png' : 'img', '.jpg' : 'img', '.jpeg': 'img', '.svg' : 'img', '.webp': 'img', #codeFile '.js' : 'txt', '.css' : 'txt', '.html' : 'txt', '.md' : 'txt', '.json' : 'txt', '.txt' : 'txt', '.py' : 'txt', '.csv' : 'txt', '.dir':'dir', } def log(msg): timestamp = datetime.now().strftime('%Y-%m-%dT%H:%M:%S') log_entry = f"{timestamp}; {msg}\n" with open(log_file, 'a') as file: file.write(log_entry) #this code use for file browser-index9 we dont use it @app.get("/{PATH0:path}") @app.post("/{PATH0:path}") async def core(reqData: Request, PATH0: str, file: UploadFile = File(None)): if PATH0 == 'favicon.ico': return '' try: route, path, tmp_path, = '', '', '' tpp = '' #queryParams = '?'+ reqData.query_params contentType = reqData.headers.get('Content-Type') host = reqData.headers.get('host') args = dict(reqData.query_params) queryParams = '?' + urlencode(reqData.query_params._dict) if args else '' path_tokens = [token for token in PATH0.replace('@','/@').split('/') if token] if contentType == 'application/json': args = await reqData.json() else : args = dict(reqData.query_params) if path_tokens: #remove host name in path_token if (path_tokens[0] == host): path_tokens.pop(0) route = path_tokens[0] path = '/'.join(path_tokens[1:]) #default route if route == '' : route='getFile'; path='index.html' # /calc.oba (run obean app ) elif route.endswith('.oba'): #route = 'getFile' if path == '' : path = 'oba/'+route+'/index.html' else : path = 'oba/'+route+'/'+path route = 'getFile' if path == 'null' or route == 'favicon.ico': return '' formData = await reqData.form() filePath = ZFS_DIR+path tmp_path = filePath print('94: ', '#'*48) print('95: headers', reqData.headers) print('95: PATH0', PATH0) print('123', formData) print('96: query :', reqData.query_params) print('97: contentType', contentType) print('98: args', args) print('99: queryParams', queryParams) print('100: path_tokens', path_tokens) print('91: path :' , path) print('92: path_tokens:', path_tokens ) print('93: route :', route ) print('95: tmp_path :', tmp_path) print('93: route :', route ) print('101: ', '#'*48) ################################################################### if route == 'ums': #url management system return JSONResponse(content={'ERROR': 'File not found'}, status_code=404) #draft mode print('157:', args) if args: return ums(args) # kTable.oba to view ums data filePath = './v2FS/oba/kTable.oba/' if not path: filePath+= 'index.html' else : filePath+= path print('157:', filePath) if os.path.isfile(filePath): return FileResponse(filePath) else: return JSONResponse(content={'ERROR': 'File not found'}, status_code=404) #progressive web app #elif route == 'pwa': #filePath = ZFS_DIR+path #print('148:', filePath ) #if os.path.isfile(filePath): # return FileResponse(filePath) #elif path_tokens[-1] in [ 'manifest.json', 'pwa.js', 'pwa_sw.js ']: # pwa.pwa(path_tokens) #else: # return JSONResponse(content={'ERROR': 'File not found'}, status_code=404) ################################################################### elif route == 'uploadFile': fileName = file.filename #zz file upload if fileName.endswith('.zx'): header = file.file.read(128) path = header.split(b'\n')[0].decode().strip() upload_dir = ZFS_DIR + path.replace('@', '/@') file_path = upload_dir+ '/default.zx' file.file.seek(0) os.makedirs(upload_dir, exist_ok=True) #file upload else: upload_dir = ZFS_DIR + formData.get('path', 'upload').replace('@', '/@') file_path = upload_dir+'/'+file.filename #check upload_dir exit if not os.path.isdir(upload_dir): return {"error": f"dir '{upload_dir}' not exit "} print('235:', upload_dir, '\n236:', file) #check max fileSize 1MB # Use shutil to copy the file from the in-memory buffer to the disk # The file.file attribute is a file-like object with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) return {"filename": file.filename, "msg": "File uploaded successfully!"} elif route == 'getFile': filePath = ZFS_DIR+path print('148:', filePath ) if os.path.isfile(filePath): return FileResponse(filePath) else: return JSONResponse(content={'ERROR': 'File not found'}, status_code=404) ################################################################## #mkdir in ZFS elif route == 'zmk': dir_path = ZFS_DIR+path.replace('@', '/@') if os.path.isdir(dir_path): return {'msg': f"Directory '{path}' already exists."} #if path.endswith('.z'): os.makedirs(dir_path, exist_ok=True) os.makedirs(dir_path, exist_ok=True) if os.path.isdir(dir_path): return {'msg': f"Directory '{path}' created successfully."} #create file in ZFS elif route == 'zcf': print('224: zcf',) file_path = ZFS_DIR+path.replace('@', '/@') if os.path.isfile(file_path): return {'msg': f"Directory '{path}' already exists."} #if path.endswith('.z'): os.makedirs(dir_path, exist_ok=True) open(file_path, 'a').close() if os.path.isfile(file_path): return {'msg': f"Directory '{path}' created successfully."} ################################################################## elif route == 'ls': #entry return with dir char '/' if os.path.isdir(tmp_path): return [_+'/' if os.path.isdir(tmp_path+'/'+_) else _ for _ in os.listdir(tmp_path)] #os.listdir(tmp_path) else : return [] elif route == 'ls.html': title = 'home' if not path_tokens else path_tokens[-1] ls = os.listdir(tmp_path) html = ls_html.ls_html( title, ls, tpp, path, path_tokens, tmp_path, args, queryParams) return PlainTextResponse(content=''.join(html), media_type="text/html") #list entries elif route == 'ls.json': print('250:', path) dir_path = ZFS_DIR + path #paths = [dir_path+'/'+entry for entry in os.listdir(dir_path)] #print('246:', paths) #entries = [] #if path_list = os.listdir(path) entries = [] for entry in os.listdir(dir_path): entry_path = dir_path+'/'+entry #entry = os.path.basename(entry_path) entry_stat = os.stat(entry_path) permissions = stat.filemode(entry_stat.st_mode) fileExt = '.dir' if os.path.isdir(entry_path) else os.path.splitext(entry)[1] entry_info = { 'title' : entry, 'type' : fileTypes.get(fileExt, ''), 'fileExt': fileExt, #|| '.file', 'size' : os.path.getsize(entry_path), 'mtime' : time.ctime(entry_stat.st_mtime), 'access' : permissions, } #check dir contain thumb for dir if fileExt == '.dir' and os.path.isfile(f'{entry_path}/thumb.jpg'): entry_info['thumb'] = 'thumb.jpg' #if entry is token than send all tokens list if entry == 't': entry_info['thread_list'] = os.listdir(entry_path) #send list file if entry == 'ls': content = '' with open(path+'/ls', 'r') as file: content = file.read() if content: entries.append({'ls':content}) entries.append(entry_info) print('105:', entry_info) return entries except Exception as e: # Catch any exceptions during the file read process raise HTTPException(status_code=500, detail=f"An error occurred while reading the file: {e}")