# 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 = '/root/ZFS/' #Path("./") #'./' log_file = './log.txt' 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 def get_ztxt(dirPath): # this code is under draft #ztxt //ztxt also send those data so sever receive less request # /zz primary info # /dir //contain all entry in thi dir # /thread.zls //contain all thread list file # /zls default zls also include # /*.zls secondary zls # /icon.* icon for z-file base64 encoding # Note:- max z-file size 1MB print('342:', dirPath) # ztxt[0]= zz # ztxt[1]= /dir # ztxt[2]= thread # ztxt[3]= *entry ztxt = ['','','',''] tmp = [] for entryName in os.listdir(dirPath): abs_path = dirPath+'/'+entryName tmp.append('\n-'+ entryName) #print('347:', abs_path, ) if os.path.isdir(abs_path): tmp.append('/'); continue elif entryName == 'zz': with open(abs_path, 'r') as file: txt = file.read() ztxt[0] = txt elif (entryName == 'zls' or entryName.endswith('..zls') ): #print('355:', abs_path) with open(abs_path, 'r') as file: txt = file.read() ztxt.append('\n/'+entryName+'\n') ztxt.append(txt) ztxt[1] = '\n/zdir\n'+''.join(tmp) return ''.join(ztxt ) @app.get("/{req_path:path}") @app.post("/{req_path:path}") async def core(reqData: Request, req_path: str, file: UploadFile = File(None)): if req_path == 'favicon.ico': return '' #if path == 'null' or route == 'favicon.ico': return '' try: contentType = reqData.headers.get('Content-Type') host = reqData.headers.get('host') ref_url = reqData.headers.get('referer') args = dict(reqData.query_params) queryParams = '?' + urlencode(reqData.query_params._dict) if args else '' path_tokens = [_ for _ in req_path.replace('@','/@').split('/') if _] formData = await reqData.form() print('94: ', '#'*48) #print('95: headers :', reqData.headers) #print('95: req_path0 :', req_path) #print('97: contentType :', contentType) #print('98: args :', args) #print('96: query :', reqData.query_params) #print('99: queryParams :', queryParams) #print('100: path_tokens :', path_tokens) #print('123: formData :', formData) #print('123: ref_url :', ref_url) #print('101: ', '#'*48) if not args and contentType == 'application/json': args = await reqData.json() if args == None : args ={} #print('137:', args) reqType, path = '','' _ = path_tokens # /ob_app.oba # /path :rx # :jl/path # # print('161:', _, req_path) if len(_) == 0 : reqType ,path = ':zob' ,'oba/zshlo.oba/index.html' elif _[0][0] == ':' : reqType ,path = _[0] ,'/'.join(_[1:]) elif _[0].endswith('.oba'): if len(_) == 1 : reqType ,path = ':rx' ,'/'.join( ['oba']+_+['index.html']) else : reqType ,path = ':rx' ,'/'.join( ['oba']+_) else : reqType ,path = ':zob' ,'/'.join(_) print('165: reqType: ', reqType) print('165: path : ', path) if reqType == ':mk': if os.path.isdir(dirPath): return {'msg': f"Directory '{path}' already exists."} #if path.endswith('.z'): os.makedirs(dir_path, exist_ok=True) os.makedirs(dirPath, exist_ok=True) if os.path.isdir(dirPath): return {'msg': f"Directory '{path}' created successfully."} #if os.path.isfile(filePath): return {'msg': f"Directory '{path}' already exists."} #if path.endswith('.z'): os.makedirs(dir_path, exist_ok=True) #open(filePath, 'a').close() #if os.path.isfile(filePath): return {'msg': f"Directory '{path}' created successfully."} elif reqType == ':rm': print('222:') elif reqType == ':rx': path = ZFS_DIR+path print('200: path: ', path) if os.path.isdir(path): print('295:', path) #ls = os.listdir(path) ls = [_+'/' if os.path.isdir(path+'/'+_) else _ for _ in os.listdir(path)] #return ls return PlainTextResponse(content='\n'.join(ls), media_type="text/html") elif os.path.isfile(path) : return FileResponse(path) elif os.path.isfile(path+'.zt'): return FileResponse(path+'.zt') elif os.path.isfile(path+'.zc'): return FileResponse(path+'.zc') elif os.path.isfile(path+'.zz'): return FileResponse(path+'.zz') elif os.path.isfile(path+'/zz'): return FileResponse(path+'/zz') else : return {'error': 'file path invalid '} elif reqType == ':wx': path = ZFS_DIR+path #if path_tokens[-1][0] == '@': path+='/zz' filePath, dirPath = path, path print('157: ZFS :- ', ztxt, path_tokens ) print('157: filePath:- ', filePath ) print('157: filePath:- ', args ) if path: upload_path = ZFS_DIR+path if not os.path.isdir(upload_path): return {"error": f"dir '{upload_path}' not exit "} else: fileName = file.filename #zz file upload if fileName.endswith('.zx'): header = file.file.read(128) path = header.split(b'\n')[0].decode().strip() upload_path= upload_dir+ '/default.zx' file.file.seek(0) os.makedirs(upload_dir, exist_ok=True) 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 if os.path.isdir(upload_path): with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) return {"filename": file.filename, "msg": "File uploaded successfully!"} elif reqType == ':zob': html_filePath = ZFS_DIR+'oba/zshlo.oba/index.html' filePath = ZFS_DIR+path ztxt = '' html_txt = 'index file not found' print('364:', html_filePath) print('365:', filePath) #readFile index.html if os.path.isfile(html_filePath): with open(html_filePath, 'r') as file: html_txt = file.read() #read .zz if os.path.isfile(filePath): with open(filePath, 'r') as file: ztxt = file.read() elif os.path.isfile(filePath+'.zz'): with open(filePath+'.zz', 'r') as file: ztxt = file.read() #read /zz elif os.path.isfile(filePath+'/zz'): ztxt = get_ztxt(filePath) print('385:', html_txt[:256]) print('386:', ztxt[:256]) #progressive web app if 'pwa' in args: tmp = '/'.join(path_tokens[1:]) head = '' body = '' html_txt = html_txt.replace('', head) html_txt = html_txt.replace('', body) #ztxt file if len(ztxt): print('387: ztxt_len', len(ztxt) ) html_txt = html_txt.replace( "", "" ) return Response(content=html_txt, media_type='text/html') elif reqType == ':ls': path = ZFS_DIR+path print('295:', path) ls = os.listdir(path) return PlainTextResponse(content="\n".join(ls), media_type="text/plain") elif path_tokens[-1] in [ 'manifest.json', 'pwa.js', 'pwa_sw.js']: print('174:---') txt, mime = pwa.pwa(path_tokens, req_path) return Response(content=txt, media_type=mime) # /zfs elif path_tokens and path_tokens[0] =='zfs': path = ZFS_DIR+'/'.join(path_tokens[1:]) #if path_tokens[-1][0] == '@': path+='/zz' filePath, dirPath = path, path print('157: ZFS :- ', ztxt, path_tokens ) print('157: filePath:- ', filePath ) print('157: filePath:- ', args ) #print('207:', os.path.isfile(filePath)) #print('207:', os.path.isdir(ZFS_DIR+'/'.join(path_tokens[1:1]))) if 'uploadFile' in args: 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 'ls.json' in args: print('250:', path) dirPath = 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(dirPath): entry_path = dirPath+'/'+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 #progresive web app if 'pwa' in args and ztxt==0: if not os.path.isfile(filePath): return JSONResponse(content={'ERROR': 'File not found'}, status_code=404) html_txt = 'index file not found' print('287:', filePath) #readFile index.html with open(filePath, 'r') as file: html_txt = file.read() head = '' body = '' html_txt = html_txt.replace('', head) html_txt = html_txt.replace('', body) return Response(content=html_txt, media_type='text/html') else: return JSONResponse(content={'ERROR': 'File not found'}, status_code=404) 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}")