import json import os,shutil,time from datetime import datetime from pathlib import Path from zipfile import ZipFile from utils.comic.ComicInfo import comicInfo from utils.Ntfy import ntfy from utils.FileUtils import fileUtils class CBZUtils: @classmethod def getCBZ_Dir(cls): return comicInfo.getNewCBZComicChapter() @classmethod def getCBZ_Path(cls): return cls.getCBZ_Dir()+".CBZ" @classmethod def PackCBZAndequZipLength(cls): try: size_zip = len(cls.zip_compression(comicInfo.getDirComicChapter(), cls.getCBZ_Path())) size_imgs = len(comicInfo.getChapterImgs()) if size_zip == size_imgs: ntfy.sendMsg(f"打包校验成功: {cls.getCBZ_Path()}") comicInfo.nextCBZToDoneChapter() return True else: ntfy.sendMsg(f"打包检验不完整:{cls.getCBZ_Path()}") return False except: print("CBZ打包ERROR") @classmethod def readDirsOrFiles(cls,dir,type): data = [] files = os.listdir(dir) for file in files: path = os.path.join(dir,file) if type == "files" and os.path.isfile(path): data.append(path) if type == "dirs" and os.path.isdir(path): data.append(path) return data @classmethod def zip_compression(cls,source_dir=None, target_file=None, remove=True): target_dir = os.path.dirname(target_file) if not os.path.exists(target_dir): os.makedirs(target_dir) if not os.path.exists(target_file) and source_dir != None: with ZipFile(target_file, mode='w') as zf: for path, dir_names, filenames in os.walk(source_dir): path = Path(path) arc_dir = path.relative_to(source_dir) y = 0 for filename in filenames: y = y + 1 print("打包中:" + str(y) + "/" + str(len(filenames)), os.path.join(source_dir, filename)) zf.write(path.joinpath(filename), arc_dir.joinpath(filename)) zf.close() ntfy.sendMsg(f"打包完成:{target_file}") else: with ZipFile(target_file, "r") as zfile: filenames = zfile.namelist() zfile.close() print("文件已存在:", target_file) if remove: filenames.remove(comicInfo.COMIC_ICON_NAME+".jpg") filenames.remove(comicInfo.COMIC_INFO_XML) return filenames @classmethod def packAutoComicChapterCBZ(cls): chapter_path = comicInfo.getDirComicChapter() if os.path.exists(chapter_path): dirs = os.listdir(chapter_path) for file in dirs: if file.startswith("scramble="): try: os.remove(file) except: print(f"删除 {file} 发生错误,已跳过") cls.zip_compression(comicInfo.getDirComicChapter(), cls.getCBZ_Path()) time.sleep(0.1) fileUtils.remove(comicInfo.getDirComicChapter()) return cls.PackCBZAndequZipLength() @classmethod def replaceZip(cls,filepath,unpack_dir=None): if not cls.compareFileDate(filepath): return None if unpack_dir == None: unpack_dir = str(filepath).split(".")[0] fz = ZipFile(filepath, 'r') for file in fz.namelist(): if file.endswith(".jpg"): data = fz.read(file) if len(data) < 500 and os.path.exists(filepath): os.remove(filepath) print(f"数据不完整,已删除:{filepath}") if cls.compareFileDate(filepath): os.utime(filepath) print(f"已更新文件时间 {filepath}") if os.path.exists(unpack_dir): shutil.rmtree(unpack_dir) # 删除删除main.ftl文件 #delete_filename = '' #if os.path.exists(delete_filename): # os.remove(delete_filename) # time.sleep(60) # shutil.copy(文件的路径,另一个目录);拷贝main.ftl到准备压缩的目录下 #cls.zip_compression() #小于则运行 @classmethod def compareFileDate(cls,filepath): if os.path.exists(filepath): ctime = os.path.getmtime(filepath) str_ctime = datetime.fromtimestamp(int(ctime)) file_ctime = str(str_ctime.year)+"{:0>2d}".format(str_ctime.month)+"{:0>2d}".format(str_ctime.day)+"{:0>2d}".format(str_ctime.hour) c_ctime = 2023011603 else: return False if int(file_ctime) < c_ctime: return True return False class verUtils: @classmethod def verNextCBZ(cls,list_img): #验证数据是已存在且是否完整 cbz_path = CBZUtils.getCBZ_Path() is_next = False if os.path.exists(cbz_path): try: cbz_size = len(CBZUtils.zip_info(cbz_path)) except: cbz_size = 0 if len(list_img) == cbz_size: ntfy.sendMsg(f"{comicInfo.getComicName()} {comicInfo.getChapter()} 数据完整,已跳过") is_next = True else: ntfy.sendMsg(f"{comicInfo.getComicName()} {comicInfo.getChapter()} 数据不完整,尝试删除配置CBZ文件后重试") try: if cbz_size < len(list_img) or os.path.getsize(cbz_path) < 300000: ntfy.sendMsg(f"删除 {cbz_path}") os.remove(cbz_path) else: is_next = True except: ntfy(f"删除失败 {cbz_path}") return is_next