Added feature for story cleaning

04-Rename-Tags.py

@@ -22,7 +22,6 @@
         tw_tags = list(csv.DictReader(csvfile))
         tag_headers = tags.tag_export_map
         total = len(tw_tags)
-
         for cur, row in enumerate(tw_tags):
             tags.update_tag_row(row)
             print_progress(cur, total, "tags")

README.md

@@ -255,6 +255,15 @@ On it we can either type (and press enter):
 Given a comma-separated list of story ids specified in the `story_ids_to_remove` parameter, deletes the corresponding
 rows from the stories table in the final output database.
 
+## Extract notes, summaries, and commercial links from chapters
+
+Extracts notes, summaries, content warnings, and commercial links from story files
+
+This is mostly for non-eFiction archives where the author's notes and content warnings are present in the main bodies of work instead of separate fields. If you use this with the 'remove_option' in the properties file populated, the script will create a backup of all works. If you hit an error or KeyboardInterrupt, the script prints out the index of the chapters list you were at. 
+
+note that this does NOT load chapters into the stories table
+
+
 
 ## Parameters
 

example.yml

@@ -28,3 +28,13 @@ bookmark_ids_to_remove: /path/to/ids to remove.txt
 # Chapter file location
 chapters_path: /path/to/chapter/files/stories
 chapters_file_extensions: txt, html
+
+
+
+# options for story cleanup
+# leave blank to ignore story files; any value here will remove flagged text from the original
+remove_option:
+chapters_backup_path: /path/to/backup/chapter/story/output/files
+output_csv: /path/to/output/Desired_CSV_name.csv
+# Comma-separated list of types to scan for
+scan_types: [notes, warnings, summary, commercial]

requirements.txt

@@ -7,7 +7,7 @@ charade==1.0.3
 Genshi==0.7.7
 ipaddr==2.2.0
 keyring==24.2.0
-lxml==4.9.3
+lxml==6.1.3
 ndg_httpsclient==0.5.1
 pyasn1==0.5.0
 pytest==7.4.3

shared_python/Args.py

@@ -269,3 +269,24 @@ def args_for_08(self):
             )
             self._print_args(self.args)
         return self.args
+
+    def args_for_sn_extraction(self):
+        if self.args.chapters_path is None:
+            self.args.chapters_path = input(
+                "Location of the text files containing the stories:"
+            )
+        if self.args.remove_option is not None:
+            if self.args.chapters_backup_path is None:
+                self.args.chapters_backup_path = input(
+                    "Desired location (full path) of the original text backup"
+                )
+            if os.path.exists(self.args.chapters_backup_path):
+                os.rmdir(self.args.chapters_backup_path)
+
+        if self.args.output_csv is None:
+            self.args.output_csv = input("Output path for the extracted CSV")
+        if self.args.scan_types is None:
+            self.args.scan_types = ["notes", "warnings", "summary", "commercial"]
+
+        self._print_args(self.args)
+        return self.args

shared_python/Common.py

@@ -1,6 +1,7 @@
 # -- coding: utf-8 --
-import sys
 from importlib import reload
+import os
+import sys
 
 reload(sys)
 # sys.setdefaultencoding('utf8') #setdefaultencoding is disabled in Python 3. UTF-8 is also default coding.
@@ -13,3 +14,19 @@ def print_progress(cur, total, prog_type="stories"):
     sys.stdout.write("\r{0}/{1} {2}".format(cur, total, prog_type))
     sys.stdout.flush()
     return cur
+
+
+def recursive_story_listdir(main_path):
+    storyfiles = [
+        x.name for x in os.scandir(main_path) if "html" in x.name or "txt" in x.name
+    ]
+    subdirs = [x.name for x in os.scandir(main_path) if x.is_dir()]
+    for dir in subdirs:
+        stories = [
+            x
+            for x in os.listdir(os.path.join(main_path, dir))
+            if "html" in x or "txt" in x
+        ]
+        for s in stories:
+            storyfiles.append(os.path.join(dir, s))
+    return storyfiles

story_cleanup/keywords.py

@@ -0,0 +1,72 @@
+class Keywords:
+    def __init__(self):
+        self.notes = [
+            "note",
+            "Note",
+            "notes",
+            "Notes",
+            "A/n",
+            "a/n",
+            "A/N",
+            "disclaimer",
+            "Disclaimer",
+            "DISCLAIMER",
+        ]
+        self.warnings = [
+            "warning",
+            "Warning",
+            "warnings",
+            "Warnings",
+            "cw",
+            "CW",
+            "content warning",
+            "Content Warning",
+            "trigger warning",
+            "Trigger Warning",
+            "Content warning",
+            "Trigger warning",
+            "tw",
+            "TW",
+        ]
+        self.summary = [
+            "summary",
+            "Summary",
+            "summaries",
+            "Summaries",
+            "prompt",
+            "Prompt",
+        ]
+        self.commercial = [
+            "Patreon",
+            "patreon",
+            "mailto",
+            "Ko-fi",
+            "ko-fi",
+            "Ko-Fi",
+            "gofundme",
+            "GoFundMe",
+        ]
+        # self.ratings
+        # self.wordcount
+        # self.pairings
+        # self.characters
+        # others defined
+
+    def detect(self, type, text):
+        match type:
+            case "notes":
+                keywords = self.notes
+            case "warnings":
+                keywords = self.warnings
+            case "summary":
+                keywords = self.summary
+            case "commercial":
+                keywords = self.commercial
+            case _:
+                raise NameError(
+                    "unsupported type: must be 'notes', 'warnings', 'summary', or 'commercial'; check config file"
+                )
+        for k in keywords:
+            if k in text:
+                return True
+        return False

story_cleanup/messages.py

@@ -0,0 +1,50 @@
+from colorama import Fore, Style
+
+
+class Messaging:
+    """
+    class that handles interactivity, message colors, etc
+    """
+
+    def __init__(self, text):
+        self.text = text
+        self.summary_color = Fore.BLUE
+        self.notes_color = Fore.MAGENTA
+        self.warning_color = Fore.LIGHTBLUE_EX
+        self.commercial_color = Fore.RED
+        self.additional_lines_color = Fore.LIGHTBLACK_EX
+        self.general = Fore.GREEN
+        self.reset = Style.RESET_ALL
+
+    def check_init_response(self, type):
+        match type:
+            case "summary":
+                print(self.summary_color + f"{type}:\n" + self.reset + self.text)
+            case "notes":
+                print(self.notes_color + f"{type}:\n" + self.reset + self.text)
+            case "warnings":
+                print(self.warning_color + f"{type}:\n" + self.reset + self.text)
+            case "commercial":
+                print(self.commercial_color + f"{type}:\n" + self.reset + self.text)
+        r = input(
+            self.general
+            + "Enter 'n' if this is not correct, otherwise press any key to continue: \n"
+            + self.reset
+        )
+        if r.lower() == "n":
+            return False
+        return True
+
+    def check_additional_lines(self, preview):
+        print(f"{self.text}\n")
+        print(self.additional_lines_color + preview + self.reset)
+        r = input(
+            (
+                self.general
+                + "Enter 'n' if you would like to add the next line to the string identified and removed, otherwise press any key to continue: \n"
+                + self.reset
+            )
+        )
+        if r == "n":
+            return True
+        return False

story_cleanup/outputs.py

@@ -0,0 +1,35 @@
+import csv
+import os
+
+
+class OutputCSV:
+    def __init__(self, filename: str, columns: list):
+        self.filename = filename
+        self.columns = columns + ["story_identifier"]
+
+    def init_csv(self):
+        with open(self.filename, "w", encoding="utf_8_sig", newline="") as fp:
+            csvf = csv.DictWriter(fp, fieldnames=self.columns)
+            csvf.writeheader()
+
+    def write_data(self, data: dict):
+        with open(self.filename, "a", encoding="utf_8_sig", newline="") as f:
+            csvf = csv.DictWriter(f, fieldnames=self.columns)
+            csvf.writerow(data)
+
+
+class OutputStoryFiles:
+    def __init__(self, output_path):
+        self.output_path = output_path
+
+    def strip_and_rewrite(self, text, options: dict):
+        text = "".join(text)
+        for o in options.keys():
+            hit_lines = options[o].split("\n")
+            for h in hit_lines:
+                text = text.replace(h, "")
+        return text
+
+    def out_to_file(self, filename, strippedtext):
+        with open(os.path.join(self.output_path, filename), "w") as f:
+            f.write(strippedtext)

story_cleanup/parsing.py

@@ -0,0 +1,41 @@
+from .keywords import Keywords
+from .messages import Messaging
+
+
+class Parser:
+    def __init__(self, textlines: list, options: list):
+        self.textlines = textlines
+        self.max = len(self.textlines)
+        self.options_dict = {}
+        for opt in options:
+            self.options_dict.update({opt: ""})
+        self.end_message = "***no more lines to preview, press enter to continue***"
+        self.kwords = Keywords()
+
+    def detection_loop(self, type: str, i: int):
+        m = Messaging(self.textlines[i])
+        if m.check_init_response(type):
+            local_count = i + 1
+            txt = self.textlines[i]
+            try:
+                a = m.check_additional_lines(self.textlines[local_count])
+            except IndexError:
+                input(self.end_message)
+                return txt
+            while a and local_count < self.max:
+                preview = self.textlines[local_count + 1]
+                txt = txt + self.textlines[local_count]
+                a = m.check_additional_lines(preview)
+                local_count += 1
+            return txt
+        return ""
+
+    def parse_lines(self):
+        # this is per story file
+        for i in range(0, int(self.max)):
+            for o in self.options_dict.keys():
+                if self.kwords.detect(o, self.textlines[i]):
+                    self.options_dict[o] = (
+                        self.options_dict[o] + "\n" + self.detection_loop(o, i)
+                    )
+        return self.options_dict

xx-Extract-Notes-Summaries-Other.py

@@ -0,0 +1,59 @@
+import os
+import shutil
+
+from shared_python.Args import Args
+from shared_python.Common import recursive_story_listdir
+
+from story_cleanup.outputs import OutputCSV, OutputStoryFiles
+from story_cleanup.parsing import Parser
+
+
+# TODO: improve logging
+"""
+For archives where authors' notes and summaries are in the main body of text, this script scans for and removes them
+note that this does /not/ load fields into the database tables, only extracts them to a CSV (and deletes if necessary)
+"""
+
+if __name__ == "__main__":
+    args_obj = Args()
+    args = args_obj.args_for_sn_extraction()
+    stories = recursive_story_listdir(args.chapters_path)
+    log = args_obj.logger_with_filename()
+
+    options = args.scan_types
+    to_remove = args.remove_option
+
+    csv = OutputCSV(args.output_csv, options)
+    j = int(
+        input("the index you left off at, if you are starting from scratch enter 0\n")
+    )
+
+    if to_remove is not None and j == 0:
+        shutil.copytree(args.chapters_path, args.chapters_backup_path)
+        rewrite = OutputStoryFiles(args.chapters_path)
+
+    try:
+        # TODO: function that queries db for chapter and filenames
+        if j == 0:
+            csv.init_csv()
+        for s in stories[j:]:
+            log.info(f"story: {s}")
+            fname = os.path.join(args.chapters_path, s)
+            with open(fname, "r") as f:
+                text = f.readlines()
+            parser = Parser(text, options)
+            hits = parser.parse_lines()
+
+            if to_remove is not None:
+                new_text = rewrite.strip_and_rewrite(text, hits)
+                rewrite.out_to_file(s, new_text)
+
+            hits.update({"story_identifier": s})
+            csv.write_data(hits)
+            j += 1
+
+    except KeyboardInterrupt:
+        print(f"progress at restart: {j}")
+    except Exception as e:
+        print(f"there's been an error: {e}\n")
+        print(f"progress at restart: {j}")