from models import db, File, User, UserPath, UploadTask from sqlalchemy import and_, or_, not_ import logging import os from datetime import datetime # --- 基礎查詢函數 --- def get_all_files(sort_by='date', sort_order='desc'): query = File.query.filter(File.filename != '.folder_marker', File.is_deleted == False) if sort_by == 'name': query = query.order_by(File.filename.asc() if sort_order == 'asc' else File.filename.desc()) elif sort_by == 'size': query = query.order_by(File.size.asc() if sort_order == 'asc' else File.size.desc()) elif sort_by == 'type': query = query.order_by(File.mime_type.asc() if sort_order == 'asc' else File.mime_type.desc()) else: query = query.order_by(File.upload_date.asc() if sort_order == 'asc' else File.upload_date.desc()) return query.all() def is_user_allowed_path(user_id, folder_path): user = User.query.get(user_id) if not user or user.username == 'admin': return True user_paths = [up.path for up in user.paths] if not user_paths: return False folder_path = folder_path.strip('/') for up in user_paths: up = up.strip('/') if folder_path == up or folder_path.startswith(up + '/'): return True return False def get_all_files_for_user(user_id, sort_by='date', sort_order='desc'): user = User.query.get(user_id) if not user: return [] if user.username == 'admin': return get_all_files(sort_by, sort_order) user_paths = [up.path for up in user.paths] if not user_paths: return [] query = File.query.filter(File.filename != '.folder_marker', File.is_deleted == False) path_filters = [File.folder.like(p + '%') for p in user_paths] query = query.filter(or_(*path_filters)) if sort_by == 'name': query = query.order_by(File.filename.asc() if sort_order == 'asc' else File.filename.desc()) elif sort_by == 'size': query = query.order_by(File.size.asc() if sort_order == 'asc' else File.size.desc()) elif sort_by == 'type': query = query.order_by(File.mime_type.asc() if sort_order == 'asc' else File.mime_type.desc()) else: query = query.order_by(File.upload_date.asc() if sort_order == 'asc' else File.upload_date.desc()) return query.all() def get_folder_contents_for_user(user_id, folder_path): folder_path = folder_path.strip('/') if not folder_path: folder_path = 'root' # 檢查是否有權限訪問該路徑 if not is_user_allowed_path(user_id, folder_path): # 即使沒有直接權限，如果是父級路徑，也允許瀏覽其子目錄（只要子目錄在允許列表裡） user = User.query.get(user_id) if not user: return None, None user_paths = [up.path.strip('/') for up in user.paths] # 檢查該 folder_path 是否是任何允許路徑的父目錄 is_parent_of_allowed = any(up.startswith(folder_path + '/') for up in user_paths) if not is_parent_of_allowed and folder_path != 'root': return None, None files = File.query.filter_by(folder=folder_path, is_deleted=False).filter(File.filename != '.folder_marker').all() subfolders_query = File.query.filter(File.filename == '.folder_marker', File.is_deleted == False) if folder_path == 'root': subfolders = subfolders_query.filter(or_(and_(not_(File.folder.contains('/')), File.folder != 'root'), File.folder.like('root/%'))).all() direct_subfolders = [] for sf in subfolders: rel = sf.folder[5:] if sf.folder.startswith('root/') else sf.folder if '/' not in rel and rel != '': direct_subfolders.append(sf) else: prefix = folder_path + "/" subfolders = subfolders_query.filter(File.folder.like(prefix + '%')).all() direct_subfolders = [sf for sf in subfolders if '/' not in sf.folder[len(prefix):]] # 再次過濾 subfolders 和 files user = User.query.get(user_id) if user and user.username != 'admin': user_paths = [up.path.strip('/') for up in user.paths] # 只保留有權限的文件 filtered_files = [f for f in files if any(f.folder == up or f.folder.startswith(up + '/') for up in user_paths)] # 資料夾過濾：如果是允許路徑，或者是允許路徑的父目錄 filtered_subfolders = [] for sf in direct_subfolders: sf_path = sf.folder.strip('/') is_allowed = any(sf_path == up or sf_path.startswith(up + '/') for up in user_paths) is_parent = any(up.startswith(sf_path + '/') for up in user_paths) if is_allowed or is_parent: filtered_subfolders.append(sf) return filtered_files, filtered_subfolders return files, direct_subfolders def search_files(query, path, file_type, min_size=None, max_size=None, start_date=None, end_date=None, sort_by='date', sort_order='desc', user_id=None): q = File.query.filter(File.is_deleted == False) # 用戶權限過濾 if user_id: user = User.query.get(user_id) if user and user.username != 'admin': user_paths = [up.path.strip('/') for up in user.paths] if not user_paths: return [] path_filters = [File.folder.like(p + '%') for p in user_paths] q = q.filter(or_(*path_filters)) if path and path != 'all' and path != 'root': q = q.filter(File.folder.like(f'{path}%')) if file_type and file_type != 'all': if file_type == 'folder': q = q.filter(File.filename == '.folder_marker') if query: q = q.filter(File.folder.like(f'%{query}%')) else: q = q.filter(File.mime_type.like(f'{file_type}%'), File.filename != '.folder_marker') if query: q = q.filter(File.filename.like(f'%{query}%')) else: if query: q = q.filter(or_( and_(File.filename != '.folder_marker', File.filename.like(f'%{query}%')), and_(File.filename == '.folder_marker', File.folder.like(f'%{query}%')) )) else: q = q.filter(File.filename != '.folder_marker') if min_size: q = q.filter(File.size >= min_size) if max_size: q = q.filter(File.size <= max_size) if start_date: q = q.filter(File.upload_date >= start_date) if end_date: q = q.filter(File.upload_date <= end_date) if sort_by == 'name': q = q.order_by(File.filename.asc() if sort_order == 'asc' else File.filename.desc()) elif sort_by == 'size': q = q.order_by(File.size.asc() if sort_order == 'asc' else File.size.desc()) elif sort_by == 'date': q = q.order_by(File.upload_date.asc() if sort_order == 'asc' else File.upload_date.desc()) else: q = q.order_by(File.upload_date.desc()) return q.all() def get_deleted_items(sort_by='date', sort_order='desc', user_id=None): query = File.query.filter_by(is_deleted=True) # 用戶權限過濾 if user_id: user = User.query.get(user_id) if user and user.username != 'admin': user_paths = [up.path.strip('/') for up in user.paths] if not user_paths: return [] path_filters = [File.folder.like(p + '%') for p in user_paths] query = query.filter(or_(*path_filters)) if sort_by == 'name': query = query.order_by(File.filename.asc() if sort_order == 'asc' else File.filename.desc()) elif sort_by == 'size': query = query.order_by(File.size.asc() if sort_order == 'asc' else File.size.desc()) elif sort_by == 'date': query = query.order_by(File.upload_date.asc() if sort_order == 'asc' else File.upload_date.desc()) else: query = query.order_by(File.upload_date.desc()) return query.all() # --- 檔案與資料夾操作 --- def get_file_by_id(db_id): return File.query.get(db_id) def add_file(filename, file_id, folder, size=None, mime_type=None, thumbnail_file_id=None, message_link=None): folder = folder.strip('/') if not folder: folder = 'root' new_file = File( filename=filename, file_id=file_id, folder=folder, size=size, mime_type=mime_type, thumbnail_file_id=thumbnail_file_id, message_link=message_link ) db.session.add(new_file); db.session.commit() logging.info(f"[DB-FLOW] Indexed: {filename} in {folder}") def _ensure_folder_path_exists(folder_path): folder_path = folder_path.strip('/') if not folder_path or folder_path == 'root': return parts = folder_path.split('/') current_path = "" for part in parts: current_path = (current_path + "/" + part) if current_path else part existing = File.query.filter_by(folder=current_path, filename='.folder_marker').first() if not existing: db.session.add(File(filename=".folder_marker", file_id="FOLDER_MARKER", folder=current_path, mime_type="application/x-folder")) db.session.commit() def rename_item(item_id, new_name, is_folder=False): item = File.query.get(item_id) if not item: return if is_folder: old_full_path = item.folder parent_path = os.path.dirname(old_full_path) new_full_path = os.path.join(parent_path, new_name).replace('\\', '/') files_to_update = File.query.filter(or_(File.folder == old_full_path, File.folder.like(old_full_path + '%'))).all() for f in files_to_update: f.folder = f.folder.replace(old_full_path, new_full_path, 1) else: item.filename = new_name db.session.commit() def restore_items(items): processed_ids = set() items_to_process = [int(item['id']) for item in items] while items_to_process: item_id = items_to_process.pop(0) if item_id in processed_ids: continue item = File.query.get(item_id) if not item: continue item.is_deleted = False processed_ids.add(item.id) if item.filename == '.folder_marker': folder_path = item.folder children = File.query.filter(or_(File.folder == folder_path, File.folder.like(f'{folder_path}/%'))).filter(File.is_deleted == True).all() for child in children: if child.id not in processed_ids: items_to_process.append(child.id) parent_path = item.folder if parent_path != 'root': parts = parent_path.split('/') current_path = '' for part in parts: current_path = f"{current_path}/{part}" if current_path else part parent_marker = File.query.filter_by(folder=current_path, filename='.folder_marker').first() if parent_marker and parent_marker.is_deleted: parent_marker.is_deleted = False; processed_ids.add(parent_marker.id) db.session.commit() def permanent_delete_items(items): for item in items: file = File.query.get(int(item['id'])) if file: db.session.delete(file) db.session.commit() def delete_item(item_ids, is_bulk=False): if not isinstance(item_ids, list): item_ids = [item_ids] for item_id in item_ids: item = File.query.get(item_id) if item: if item.filename == '.folder_marker': folder_path = item.folder File.query.filter(or_(File.folder == folder_path, File.folder.like(f'{folder_path}/%'))).update({File.is_deleted: True}, synchronize_session=False) else: item.is_deleted = True db.session.commit() def create_folder(folder_path): folder_path = folder_path.strip('/') if not folder_path: return root_marker = File.query.filter_by(folder='root', filename='.folder_marker').first() if not root_marker: db.session.add(File(filename=".folder_marker", file_id="FOLDER_MARKER", folder='root', mime_type="application/x-folder")) if folder_path == 'root': db.session.commit(); return path_parts = folder_path.split('/') current_path = '' for part in path_parts: current_path = f"{current_path}/{part}" if current_path else part existing = File.query.filter_by(folder=current_path, filename='.folder_marker').first() if not existing: db.session.add(File(filename=".folder_marker", file_id="FOLDER_MARKER", folder=current_path, mime_type="application/x-folder")) db.session.commit() def move_items(items, destination_folder): destination_folder = destination_folder.strip('/') if not destination_folder: destination_folder = 'root' # 強制修正 root/ 前綴，確保移動後的路徑可追蹤 if not destination_folder.startswith('root') and destination_folder != 'root': destination_folder = 'root/' + destination_folder if not File.query.filter_by(folder=destination_folder, filename='.folder_marker').first() and destination_folder != 'root': _ensure_folder_path_exists(destination_folder) for item_data in items: item = File.query.get(item_data['id']) if not item: continue if item_data['type'] == 'file': item.folder = destination_folder elif item_data['type'] == 'folder': old_folder_path = item.folder folder_name = os.path.basename(old_folder_path) new_folder_path = os.path.join(destination_folder, folder_name).replace('\\', '/') files_to_move = File.query.filter(File.folder.like(old_folder_path + '%')).all() for f in files_to_move: f.folder = f.folder.replace(old_folder_path, new_folder_path, 1) db.session.commit() def copy_items(items, destination_folder): destination_folder = destination_folder.strip('/') if not destination_folder: destination_folder = 'root' # 強制修正 root/ 前綴 if not destination_folder.startswith('root') and destination_folder != 'root': destination_folder = 'root/' + destination_folder if not File.query.filter_by(folder=destination_folder, filename='.folder_marker').first() and destination_folder != 'root': _ensure_folder_path_exists(destination_folder) for item_data in items: item = File.query.get(item_data['id']) if not item: continue if item_data['type'] == 'file': db.session.add(File(filename=item.filename, file_id=item.file_id, folder=destination_folder, size=item.size, mime_type=item.mime_type, thumbnail_file_id=item.thumbnail_file_id, message_link=item.message_link)) elif item_data['type'] == 'folder': old_folder_path = item.folder folder_name = os.path.basename(old_folder_path) new_folder_path = os.path.join(destination_folder, folder_name).replace('\\', '/') items_to_copy = File.query.filter(File.folder.like(old_folder_path + '%')).all() for itc in items_to_copy: new_item_folder = itc.folder.replace(old_folder_path, new_folder_path, 1) db.session.add(File(filename=itc.filename, file_id=itc.file_id, folder=new_item_folder, size=itc.size, mime_type=itc.mime_type, thumbnail_file_id=itc.thumbnail_file_id, message_link=itc.message_link)) db.session.commit() def bulk_rename_items(items, rename_method, new_name=None, rename_template=None, find_string=None, replace_string=None): for i, item_data in enumerate(items): item = File.query.get(item_data['id']) if not item: continue original_name = os.path.basename(item.folder) if item_data['type'] == 'folder' else item.filename name, ext = os.path.splitext(original_name) final_name = "" if rename_method == 'new_name': # 確保補上副檔名（如果是文件） if item_data['type'] == 'file' and not new_name.endswith(ext): final_name = new_name + ext else: final_name = new_name elif rename_method == 'template': # 支援多種佔位符，並確保保留 ext final_name = rename_template.replace('{i}', str(i+1)).replace('{name}', name).replace('{ext}', ext).replace('[original_name]', name).replace('[extension]', ext).replace('[counter]', str(i+1)) if item_data['type'] == 'file' and not final_name.endswith(ext): final_name += ext elif rename_method == 'find_replace': final_name = original_name.replace(find_string, replace_string) else: continue if not final_name or final_name == ext: continue # 防止空白名稱 if item_data['type'] == 'folder': old_full_path = item.folder parent_path = os.path.dirname(old_full_path) new_full_path = os.path.join(parent_path, final_name).replace('\\', '/') files_to_update = File.query.filter(File.folder.like(old_full_path + '%')).all() for f in files_to_update: f.folder = f.folder.replace(old_full_path, new_full_path, 1) else: item.filename = final_name db.session.commit() def get_subfolders_by_path(parent_path): """資料夾樹專用：獲取直接子資料夾標記""" parent_path = parent_path.strip('/') if not parent_path: parent_path = 'root' like_path = parent_path + "/%" all_markers = File.query.filter(File.folder.like(like_path), File.filename == '.folder_marker', File.is_deleted == False).all() return [m for m in all_markers if os.path.dirname(m.folder) == parent_path] def get_db_id_by_file_id(file_id): file = File.query.filter_by(file_id=file_id, is_deleted=False).first() return file.id if file else None def get_filename_by_telegram_file_id(telegram_file_id, include_deleted=False): file = get_file_by_telegram_file_id(telegram_file_id, include_deleted) return file.filename if file else None def get_file_by_telegram_file_id(telegram_file_id, include_deleted=False): query = File.query.filter_by(file_id=telegram_file_id) if not include_deleted: query = query.filter_by(is_deleted=False) return query.first() def get_file_by_telegram_file_id_including_deleted(telegram_file_id): return File.query.filter_by(file_id=telegram_file_id).first() def is_file_deleted(telegram_file_id): file = get_file_by_telegram_file_id(telegram_file_id, include_deleted=True) return file.is_deleted if file else False def get_root_folder_id(): file = File.query.filter_by(folder='root', filename='.folder_marker', is_deleted=False).first() return file.id if file else None def get_folder_contents(current_folder): files = File.query.filter_by(folder=current_folder, is_deleted=False).filter(File.filename != '.folder_marker').all() like_path = 'root/%' if current_folder == 'root' else current_folder + '/%' all_subfolders = File.query.filter(File.folder.like(like_path), File.filename == '.folder_marker', File.is_deleted == False).all() direct_subfolders = [sub for sub in all_subfolders if os.path.dirname(sub.folder) == current_folder] return files, direct_subfolders def empty_recycle_bin(): File.query.filter_by(is_deleted=True).delete(); db.session.commit() def get_upload_task_by_filename(filename): return UploadTask.query.filter_by(filename=filename).order_by(UploadTask.created_at.desc()).first() def get_upload_task_by_temp_path(temp_path): return UploadTask.query.filter_by(temp_file_path=temp_path).first() def bulk_delete_upload_tasks(task_ids): UploadTask.query.filter(UploadTask.id.in_(task_ids)).delete(synchronize_session=False); db.session.commit() def get_files_by_folder(folder): return File.query.filter_by(folder=folder, is_deleted=False).all() def get_all_folders(): return [r[0] for r in db.session.query(File.folder).filter(File.is_deleted == False).distinct()] # --- 上傳任務管理 --- def get_upload_task_by_id(task_id): return UploadTask.query.get(task_id) def add_upload_task_to_db(filename, temp_file_path, target_folder, file_size, mime_type, api_used, status='queued', priority=0, thumbnail_path=None): target_folder = target_folder.strip('/') if not target_folder: target_folder = 'root' new_task = UploadTask( filename=filename, temp_file_path=temp_file_path, target_folder=target_folder, file_size=file_size, mime_type=mime_type, api_used=api_used, status=status, priority=priority, thumbnail_path=thumbnail_path ) db.session.add(new_task); db.session.commit() logging.info(f"[DB-FLOW] Task {new_task.id} Queued: {filename}") def get_upload_tasks(status=None, limit=None, sort_by='created_at', sort_order='desc'): query = UploadTask.query if status: query = query.filter_by(status=status) # 動態排序 attr = getattr(UploadTask, sort_by, UploadTask.created_at) if sort_order == 'desc': query = query.order_by(attr.desc()) else: query = query.order_by(attr.asc()) if limit: query = query.limit(limit) return query.all() def update_upload_task_status(task_id, status, error_message=None): task = UploadTask.query.get(task_id) if task: task.status = status if error_message: task.error_message = error_message db.session.commit() def bulk_update_upload_task_status(task_ids, status, error_message=None): UploadTask.query.filter(UploadTask.id.in_(task_ids)).update({UploadTask.status: status, UploadTask.error_message: error_message}, synchronize_session=False) db.session.commit() def get_folder_id_by_path(path): path = path.strip('/') if not path: path = 'root' folder = File.query.filter_by(folder=path, filename='.folder_marker').first() return folder.id if folder else None def get_all_files_in_folder(folder_path): folder_path = folder_path.strip('/') return File.query.filter_by(folder=folder_path, is_deleted=False).all() # --- 用戶與管理 --- def add_user(username, password): if User.query.filter_by(username=username).first(): return None new_user = User(username=username, password=password) db.session.add(new_user); db.session.commit(); return new_user def get_user_by_id(user_id): return User.query.get(user_id) def get_user_by_username(username): return User.query.filter_by(username=username).first() def get_all_users(): return User.query.all() def delete_user(user_id): user = User.query.get(user_id); if user: db.session.delete(user); db.session.commit() def update_user_paths(user_id, paths): UserPath.query.filter_by(user_id=user_id).delete() for p in paths: if p.strip(): db.session.add(UserPath(user_id=user_id, path=p.strip())) db.session.commit() def get_user_paths(user_id): return [p.path for p in UserPath.query.filter_by(user_id=user_id).all()] def clear_database(): File.query.delete(); UploadTask.query.delete(); db.session.commit() def update_upload_task_priority(task_id, priority): task = UploadTask.query.get(task_id) if task: task.priority = priority; db.session.commit() def delete_upload_task(task_id): task = UploadTask.query.get(task_id) if task: db.session.delete(task); db.session.commit()