mirror of
https://codeberg.org/Cyberpro123/AO3-DL.git
synced 2026-11-01 03:51:01 +00:00
363 lines
13 KiB
Python
Executable file
363 lines
13 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# import concurrent.futures
|
|
import datetime
|
|
import logging
|
|
import os.path
|
|
import platform
|
|
import shutil
|
|
import traceback
|
|
|
|
import bs4
|
|
|
|
import constants
|
|
import db
|
|
import init
|
|
|
|
|
|
def tagToFile(
|
|
tag,
|
|
filename: str,
|
|
logger: logging.Logger,
|
|
) -> None:
|
|
if isinstance(tag, str):
|
|
tagStr = tag
|
|
if isinstance(tag, list):
|
|
tagStr = ""
|
|
for i in tag:
|
|
try:
|
|
tagStr += i.prettify(formatter="html5")
|
|
except AttributeError:
|
|
logger.log(
|
|
int(30 - (20 * int(isinstance(i, bs4.NavigableString)))),
|
|
f"tagToFile() encountered [{type(i)}] which doesn't have a 'prettify' method",
|
|
)
|
|
tagStr += str(i)
|
|
if isinstance(tag, bs4.Tag):
|
|
tagStr = tag.prettify(formatter="html5")
|
|
with open(filename, "w") as file:
|
|
file.write(tagStr)
|
|
|
|
|
|
def processSingle(
|
|
workID: int,
|
|
config: dict,
|
|
logger: logging.Logger,
|
|
) -> None:
|
|
logger.info(f"Started Processing [{workID}]")
|
|
sqlInfo = {}
|
|
rawFilename = os.path.join(config["dirRaws"], f"{workID}.html")
|
|
with open(rawFilename) as rawFile:
|
|
rawSoup = bs4.BeautifulSoup(rawFile, features="lxml")
|
|
wIdFolder = os.path.join(config["dirOut"], "HTML", f"{workID}")
|
|
if not os.path.exists(wIdFolder):
|
|
os.mkdir(wIdFolder)
|
|
if not os.path.isdir(wIdFolder):
|
|
raise constants.InvalidConfigurationError(
|
|
f"Given filepath [{wIdFolder}] expected to be a directory, isn't."
|
|
)
|
|
|
|
################################################################
|
|
# Tag Isolation
|
|
################################################################
|
|
################################
|
|
# TagsDiv
|
|
################################
|
|
headerMeta = rawSoup.find(id="preface").div
|
|
tagsDiv = headerMeta.dl
|
|
tagsDiv["class"] = ["works meta group"]
|
|
workTags = tagsDiv.find_all("dd")
|
|
workTagHeadersRaw = tagsDiv.find_all("dt")
|
|
workTagHeaders = []
|
|
for tag in workTagHeadersRaw:
|
|
tagStr = tag.string
|
|
if tagStr == "Categories:":
|
|
tagStr = "Category:"
|
|
if tagStr[-2:] == "s:":
|
|
tagStr = f"{tagStr[:-2]}:"
|
|
workTagHeaders.append(tagStr)
|
|
del workTagHeadersRaw
|
|
sqlInfo["rating"] = workTags[workTagHeaders.index("Rating:")].a.string
|
|
sqlInfo["warnings"] = []
|
|
sqlInfo["categories"] = []
|
|
sqlInfo["tagsFandom"] = []
|
|
sqlInfo["tagsShips"] = []
|
|
sqlInfo["tagsChara"] = []
|
|
sqlInfo["tagsOther"] = []
|
|
for i in (
|
|
("warnings", "Archive Warning:", "Warnings!?!?!"),
|
|
("categories", "Category:", "Categories"),
|
|
("tagsFandom", "Fandom:", "Fandom Tags?!?!"),
|
|
("tagsShips", "Relationship:", "Ship Tags"),
|
|
("tagsChara", "Character:", "Character Tags"),
|
|
("tagsOther", "Additional Tag:", "General Tags"),
|
|
):
|
|
try:
|
|
for tag in workTags[workTagHeaders.index(i[1])].find_all("a"):
|
|
sqlInfo[i[0]].append(tag.string)
|
|
except ValueError:
|
|
logger.debug(f"Work [{workID}] doesn't seem to have any [{i[2]}]")
|
|
seriesList = []
|
|
for i in workTags[workTagHeaders.index("Serie:")]:
|
|
seriesList.append(i)
|
|
sqlInfo["serieses"] = []
|
|
for num, tag in enumerate(seriesList):
|
|
if isinstance(tag, bs4.element.Tag) and tag.name == "a":
|
|
hrefSplit = tag["href"].split("/")
|
|
seriesID = int(hrefSplit[1 + hrefSplit.index("series")])
|
|
for char in str(seriesList[num - 1]):
|
|
if char.isdigit():
|
|
seriesPos = int(char)
|
|
sqlInfo["serieses"].append((seriesID, seriesPos))
|
|
del seriesList, hrefSplit, seriesID, seriesPos
|
|
statsSplit = workTags[workTagHeaders.index("Stat:")].string.split()
|
|
chapterNOs = statsSplit[statsSplit.index("Chapters:") + 1].split("/")
|
|
sqlInfo["nchapters"] = int(chapterNOs[0])
|
|
try:
|
|
sqlInfo["chaptersExpected"] = int(chapterNOs[1])
|
|
except ValueError:
|
|
sqlInfo["chaptersExpected"] = 0
|
|
del chapterNOs
|
|
sqlInfo["dateUp"] = 0
|
|
for i in ("Completed:", "Updated:", "Published:"):
|
|
if not sqlInfo["dateUp"]:
|
|
try:
|
|
sqlInfo["dateUp"] = int(
|
|
datetime.datetime.fromisoformat(
|
|
statsSplit[1 + statsSplit.index(i)]
|
|
).timestamp()
|
|
)
|
|
except ValueError:
|
|
pass
|
|
if not sqlInfo["dateUp"]:
|
|
logger.error(f"Date Updated couldn't be found in work [{workID}]")
|
|
sqlInfo["datePb"] = int(
|
|
datetime.datetime.fromisoformat(
|
|
statsSplit[1 + statsSplit.index("Published:")]
|
|
).timestamp()
|
|
)
|
|
allComments = rawSoup.find_all(string=lambda text: isinstance(text, bs4.Comment))
|
|
sqlInfo["dateDl"] = 0
|
|
sqlInfo["dateEd"] = 0
|
|
for comment in allComments:
|
|
if constants.commentHeader in str(comment):
|
|
commentSplit = str(comment).splitlines()
|
|
for i in (
|
|
(constants.commentTimestampDownloaded, "dateDl"),
|
|
(constants.commentTimestampEdited, "dateEd"),
|
|
):
|
|
for line in commentSplit:
|
|
if i[0] in line:
|
|
sqlInfo[i[1]] = int(
|
|
datetime.datetime.fromisoformat(
|
|
line[len(i[0]):]
|
|
).timestamp()
|
|
)
|
|
if not sqlInfo["dateDl"]:
|
|
logger.info(
|
|
f"Raw [{workID}] doesn't seem to have a 'last downloaded' timestamp"
|
|
)
|
|
if not sqlInfo["dateEd"]:
|
|
logger.info(f"Raw [{workID}] doesn't seem to have a 'last edited' timestamp")
|
|
tagToFile(
|
|
tag=tagsDiv,
|
|
filename=os.path.join(wIdFolder, "tags.html"),
|
|
logger=logger,
|
|
)
|
|
del tagsDiv
|
|
################################
|
|
# Rest of Header
|
|
################################
|
|
sqlInfo["title"] = headerMeta.h1.string
|
|
headerBlockquotes = headerMeta.find_all("blockquote", class_="userstuff")
|
|
match len(headerBlockquotes):
|
|
case 2:
|
|
tagToFile(
|
|
tag=headerBlockquotes[1].contents,
|
|
filename=os.path.join(wIdFolder, "start-notes.html"),
|
|
logger=logger,
|
|
)
|
|
tagToFile(
|
|
tag=headerBlockquotes[0].contents,
|
|
filename=os.path.join(wIdFolder, "summary.html"),
|
|
logger=logger,
|
|
)
|
|
case 1:
|
|
match headerMeta.p.string:
|
|
case "Summary":
|
|
hbqDest = "summary.html"
|
|
case "Notes":
|
|
hbqDest = "start-notes.html"
|
|
tagToFile(
|
|
tag=headerBlockquotes[0].contents,
|
|
filename=os.path.join(wIdFolder, hbqDest),
|
|
logger=logger,
|
|
)
|
|
del hbqDest
|
|
case 0:
|
|
logger.debug(
|
|
f"Work [{workID}] seems to have neither a summary nor start notes."
|
|
)
|
|
sqlInfo["authors"] = []
|
|
for tag in headerMeta.div.find_all("a"):
|
|
paraOpenLoc = tag.string.rfind("(")
|
|
paraCloseLoc = tag.string.rfind(")")
|
|
if paraOpenLoc < paraCloseLoc and paraOpenLoc != -1 and paraCloseLoc != -1:
|
|
# pseud detected
|
|
sqlInfo["authors"].append(tag.string[paraOpenLoc + 1: paraCloseLoc])
|
|
else:
|
|
# not a pseud
|
|
sqlInfo["authors"].append(tag.string)
|
|
del paraOpenLoc
|
|
del paraCloseLoc
|
|
try:
|
|
tagToFile(
|
|
tag=rawSoup.find(id="afterword").find(id="endnotes").blockquote.contents,
|
|
filename=os.path.join(wIdFolder, "end-notes.html"),
|
|
logger=logger,
|
|
)
|
|
except AttributeError:
|
|
logger.debug(f"Work [{workID}] doesn't seem to have work end notes")
|
|
allPrefaces = rawSoup.select("#chapters > div.meta.group")
|
|
allMains = rawSoup.select("#chapters > div.userstuff")
|
|
allPostfaces = {}
|
|
for i in range(1, 1 + len(allMains)):
|
|
try:
|
|
allPostfaces[i] = rawSoup.find(id=f"endnotes{i}")
|
|
except KeyError:
|
|
logger.debug(f"Work [{workID}] Chapter [{i}] doesn't seem to have endnotes")
|
|
logger.debug(
|
|
f"Pref/Main/Post for [{workID}]: [{len(allPrefaces)}] [{len(allMains)}] [{len(allPostfaces)}]"
|
|
)
|
|
for chaptNum in range(1, 1 + len(allMains)):
|
|
chaptFolder = os.path.join(wIdFolder, str(chaptNum))
|
|
os.makedirs(chaptFolder, exist_ok=True)
|
|
################################
|
|
# Real
|
|
################################
|
|
if len(allPrefaces) != 0:
|
|
tagToFile(
|
|
tag=allPrefaces[chaptNum - 1].h2.string,
|
|
filename=os.path.join(chaptFolder, "title.txt"),
|
|
logger=logger,
|
|
)
|
|
preBQs = allPrefaces[chaptNum - 1].find_all("blockquote")
|
|
if len(preBQs) == 2:
|
|
tagToFile(
|
|
tag=preBQs[0].contents,
|
|
filename=os.path.join(chaptFolder, "summary.html"),
|
|
logger=logger,
|
|
)
|
|
tagToFile(
|
|
tag=preBQs[1].contents,
|
|
filename=os.path.join(chaptFolder, "notes-start.html"),
|
|
logger=logger,
|
|
)
|
|
elif len(preBQs) == 1:
|
|
pCandidates = allPrefaces[chaptNum - 1].select("div.meta.group > p")
|
|
match pCandidates[-1].string:
|
|
case "Chapter Notes":
|
|
bqDest = "notes-start.html"
|
|
case "Chapter Summary":
|
|
bqDest = "summary.html"
|
|
tagToFile(
|
|
tag=preBQs[0].contents,
|
|
filename=os.path.join(chaptFolder, bqDest),
|
|
logger=logger,
|
|
)
|
|
del pCandidates
|
|
del bqDest
|
|
del preBQs
|
|
elif len(allMains) == 1 and sqlInfo["nchapters"] == 1:
|
|
logger.debug(f"Work [{workID}] seems to be a oneshot")
|
|
else:
|
|
logger.error(
|
|
f"Work [{workID}], which doesn't seem to be a oneshot, had a len(allPrefaces) of 0"
|
|
)
|
|
tagToFile(
|
|
tag=allMains[chaptNum - 1].contents,
|
|
filename=os.path.join(chaptFolder, "main.html"),
|
|
logger=logger,
|
|
)
|
|
tail = allPostfaces.get(chaptNum)
|
|
if tail:
|
|
tagToFile(
|
|
tag=tail.blockquote.contents,
|
|
filename=os.path.join(chaptFolder, "notes-end.html"),
|
|
logger=logger,
|
|
)
|
|
|
|
db.addWork(
|
|
id=workID,
|
|
info=sqlInfo,
|
|
config=config,
|
|
logger=logger,
|
|
)
|
|
if os.path.exists(os.path.join(config["dirRaws"], f"{workID}.css")):
|
|
shutil.copy(
|
|
src=os.path.join(config["dirRaws"], f"{workID}.css"),
|
|
dst=os.path.join(wIdFolder, "workskin.css"),
|
|
)
|
|
logger.info(f"Completed Processing [{workID}]")
|
|
|
|
|
|
def multithreading(
|
|
workIDs: set,
|
|
config: dict,
|
|
logger: logging.Logger,
|
|
) -> None:
|
|
if (len(workIDs) < 10) or (platform.system() == "Windows"):
|
|
for workID in workIDs:
|
|
processSingle(
|
|
workID=workID,
|
|
config=config,
|
|
logger=logger,
|
|
)
|
|
else:
|
|
import concurrent.futures
|
|
|
|
futures = {}
|
|
with concurrent.futures.ThreadPoolExecutor(
|
|
max_workers=10, thread_name_prefix=constants.threadNameBulk
|
|
) as pool:
|
|
for workID in workIDs:
|
|
futures[workID] = pool.submit(
|
|
processSingle, workID=workID, config=config, logger=logger
|
|
)
|
|
logger.info("Mostly complete, getting exceptions from futures...")
|
|
for workID in futures:
|
|
try:
|
|
futures[workID].result()
|
|
except FileNotFoundError as ex:
|
|
split = str(ex).split()
|
|
string = ""
|
|
for num, i in enumerate(split):
|
|
if num == 7:
|
|
string += i
|
|
elif num > 7:
|
|
string += f" {i}"
|
|
init.logger.warning(f"Work [{workID}] couldn't find file [{string}]")
|
|
del split
|
|
del string
|
|
except Exception as ex:
|
|
init.logger.error(
|
|
f"Work [{workID}] raised [{type(ex).__name__}]: [{ex.args}]"
|
|
)
|
|
traceback.print_exception(ex)
|
|
logger.info("Complete, process.multithreding exiting")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
init.init()
|
|
workIDs = set(())
|
|
inRaw = init.parseInfile(init.args.infile)
|
|
for line in inRaw:
|
|
try:
|
|
workIDs.add(int(line))
|
|
except ValueError:
|
|
init.logger.error(f"Could not convert input [{line}] to int")
|
|
|
|
multithreading(
|
|
workIDs=workIDs,
|
|
config=init.config,
|
|
logger=init.logger,
|
|
)
|