181 lines
7.0 KiB
Python
181 lines
7.0 KiB
Python
import os,shutil,time,requests
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from zipfile import ZipFile
|
||
from common.ComicInfo import ComicInfoUtils as ciUtils
|
||
from common.ComicInfo import ComicInfo as ci
|
||
from common.ComicInfo import Comic
|
||
from utils.FileUtils import fileUtils as fu
|
||
from common.Constant import pathStr
|
||
from common.Constant import ComicPath
|
||
from utils.OldUtils import OldUtils
|
||
|
||
class ntfy:
|
||
@classmethod
|
||
def sendMsg(cls, msg,alert=True,sleep=None,error=None):
|
||
try:
|
||
print(f"#ntfy: {msg}")
|
||
if alert:
|
||
requests.post("https://ntfy.caiwenxiu.cn/PyComic",
|
||
data=msg.encode(encoding='utf-8'))
|
||
except:
|
||
print(f"#ntfy error: {msg}")
|
||
if sleep != None:
|
||
time.sleep(int(sleep))
|
||
if error != None:
|
||
print(f"#ntfy Error: {error}")
|
||
return False
|
||
else:
|
||
return True
|
||
class CBZUtils:
|
||
|
||
@classmethod
|
||
def getCBZ_Dir(cls): return ComicPath.getNewCBZComicChapter()
|
||
|
||
@classmethod
|
||
def getCBZ_Path(cls): return ComicPath.getNewFileCBZComicChapter()
|
||
|
||
@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}")
|
||
cls.verCBZComic(target_file)
|
||
|
||
@classmethod
|
||
def packAutoComicChapterCBZ(cls):
|
||
chapter_path = ComicPath.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} 发生错误，已跳过")
|
||
return False
|
||
cls.zip_compression(ComicPath.getDirComicChapter(), cls.getCBZ_Path())
|
||
time.sleep(0.1)
|
||
fu.remove(ComicPath.getDirComicChapter())
|
||
return True
|
||
|
||
@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)
|
||
|
||
@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
|
||
|
||
@classmethod
|
||
def zip_info(cls,path,filter=True):
|
||
result = None
|
||
try:
|
||
with ZipFile(path, "r") as zip_file:
|
||
result = zip_file.namelist()
|
||
if filter:
|
||
filter_icon = ci.COMIC_ICON_NAME+".jpg"
|
||
filter_info_xml = ci.COMIC_INFO_XML
|
||
if filter_icon in result: result.remove(filter_icon)
|
||
if filter_info_xml in result: result.remove(filter_info_xml)
|
||
except Exception as e:
|
||
print(e)
|
||
return result
|
||
|
||
#CBZ检验是否完整
|
||
@classmethod
|
||
def verCBZComic(cls,path=None,list_img=None,min_size=300000):
|
||
#数据检验
|
||
if path == None: path = cls.getCBZ_Path()
|
||
#文件不存在 则返回
|
||
if fu.notExists(path): return False
|
||
if list_img == None: list_img = Comic.getChapterImgs()
|
||
|
||
if fu.exists(path) and len(cls.zip_info(path)) == len(list_img):
|
||
print(f"文件校验成功：{path}")
|
||
ciUtils.setProgress(ciUtils.PROGRESS_DONE)
|
||
return True
|
||
else:
|
||
try:
|
||
if len(cls.zip_info(path)) < len(list_img) or os.path.getsize(path) < min_size:
|
||
fu.remove(path)
|
||
ciUtils.setProgress(ciUtils.PROGRESS_NONE)
|
||
except Exception as e:
|
||
print(e)
|
||
return False
|
||
|
||
@classmethod
|
||
def updateOldCBZ(cls,filesname,result=False):
|
||
old_zipfile_path = ComicPath.setJoinPathDir([OldUtils.getOldComicName(),OldUtils.getOldChapter()],
|
||
pathStr.old_cbz_path,prefix="CBZ")
|
||
#判断是否存在已下载CBZ文件
|
||
if fu.exists(old_zipfile_path) and fu.notExists(CBZUtils.getCBZ_Path()):
|
||
print(f"存在CBZ文件{old_zipfile_path},解压中...")
|
||
zip_file = ZipFile(old_zipfile_path)
|
||
#CBZ中文件数量，剔除ComicInfo.xml
|
||
if len(filesname) == len(zip_file.namelist())-1:
|
||
unzip_path = ComicPath.getDirComicChapter()
|
||
zip_file.extractall(unzip_path)
|
||
zip_file.close()
|
||
print(f"解压完成: CBZ文件{old_zipfile_path}")
|
||
print("文件校验中...")
|
||
for file in os.listdir(unzip_path):
|
||
#检验图片损坏则删除
|
||
if file.endswith(".jpg") and not fu.ver_file(os.path.join(unzip_path,file),type="image"):
|
||
fu.remove(unzip_path)
|
||
return False
|
||
ci.writeComicInfoXML(overlay=True)
|
||
result = True
|
||
return result
|
||
|
||
@classmethod
|
||
def nextCBZ(cls,list_img=None):
|
||
if list_img == None: list_img = Comic.getChapterImgs()
|
||
return not cls.verCBZComic(list_img=list_img) |