OD_177

README.md

@@ -255,6 +255,12 @@ 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.
 
+### Test links in story files
+Note: at the very least, requires links in the story files to be formatted as html, as it uses BeautifulSoup and html parsing to detect links
+
+given chapters filepath and an output file path (which you can specify using the .yml file), writes a csv with each non-relative link, the returned status or error, and the status codes of any automatic redirections
+
+the script prompts the user for the chapter file encoding, so make sure you have run the mojibake tool beforehand!
 
 ## Parameters
 

shared_python/Args.py

@@ -264,3 +264,19 @@ def args_for_08(self):
             )
             self._print_args(self.args)
         return self.args
+
+    def args_for_link_check(self):
+        if self.args.chapters_path is None:
+            self.args.chapters_path = input(
+                "Location of the text files containing the stories:"
+            )
+        if not os.path.exists(self.args.output_folder):
+            os.makedirs(self.args.output_folder)
+        while self.args.output_folder is None or not os.path.isdir(
+            self.args.output_folder
+        ):
+            self.args.output_folder = input("Path for output files: ")
+            if not os.path.exists(self.args.output_folder):
+                os.makedirs(self.args.output_folder)
+        self._print_args(self.args)
+        return self.args

shared_python/Common.py

@@ -12,3 +12,19 @@ def print_progress(cur, total, prog_type="stories"):
     sys.stdout.write(f"\r{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

shared_python/Parsers.py

@@ -0,0 +1,26 @@
+from bs4 import BeautifulSoup as bs
+
+
+class LinkParser:
+    def __init__(self, filetext):
+        self.filetext = filetext
+
+    def _is_relative(self, link):
+        if link.startswith(".") or link.startswith("/"):
+            return True
+        elif not link.startswith("http"):
+            return True
+        else:
+            return False
+
+    def extract_links(self):
+        soup = bs(self.filetext, "html.parser")
+        links = [
+            l.get("href")
+            for l in soup.find_all("a", href=True)
+            if not self._is_relative(l.get("href"))
+        ]
+        for i in soup.find_all("img"):
+            if not self._is_relative(i.get("src")):
+                links.append(i.get("src"))
+        return links

shared_python/Requests.py

@@ -0,0 +1,40 @@
+import requests
+
+
+class RequestHandler:
+    def __init__(self):
+        self.headers = {
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:92.0) Gecko/20100101 Firefox/92.0"
+        }
+        self.redirect_codes = [301, 302, 307, 308]
+
+    def _send_head(self, url):
+        try:
+            r = requests.head(url=url, headers=self.headers, timeout=10)
+        except requests.exceptions.Timeout:
+            r = "request timeout"
+        except Exception as e:
+            r = str(e)
+        return r
+
+    def status_code_handler(self, url):
+        return_dict = {
+            "status_code": "",
+            "redirect_location": "",
+            "redirect_response": "",
+        }
+        status = self._send_head(url)
+        if isinstance(status, str):
+            return_dict.update({"status_code": status})
+        else:
+            if status.status_code in self.redirect_codes:
+                return_dict.update({"redirect_location": status.headers["location"]})
+                redirect_status = self._send_head(status.headers["location"])
+                if isinstance(redirect_status, str):
+                    return_dict.update({"redirect_response": redirect_status})
+                else:
+                    return_dict.update(
+                        {"redirect_response": redirect_status.status_code}
+                    )
+            return_dict.update({"status_code": status.status_code})
+        return return_dict

xx-Test-Story-Content-Links.py

@@ -0,0 +1,70 @@
+import csv
+import os
+
+from shared_python.Args import Args
+from shared_python.Common import print_progress, recursive_story_listdir
+from shared_python.Logging import logger
+from shared_python.Parsers import LinkParser
+from shared_python.Requests import RequestHandler
+
+
+args_obj = Args()
+args = args_obj.args_for_link_check()
+log = args_obj.logger_with_filename()
+handler = RequestHandler()
+
+story_files_folder = args.chapters_path
+output_folder = args.output_folder
+file_encoding = input(
+    "Enter encoding of the story files (leave blank for utf-8):\n"
+).lower()
+if file_encoding == "":
+    file_encoding = "utf-8"
+
+stories = recursive_story_listdir(story_files_folder)
+log.info(f"Preparing to parse through {len(stories)} stories")
+
+with open(f"{output_folder}/story_file_links.csv", "a") as f:
+    writer = csv.DictWriter(
+        f,
+        fieldnames=[
+            "story_id",
+            "link",
+            "response",
+            "redirect_location",
+            "redirect_response",
+        ],
+    )
+    cur = 0
+    for s in stories:
+        try:
+            with open(
+                f"{story_files_folder}/{s}", mode="r", encoding=file_encoding
+            ) as f:
+                story_text = f.read()
+            parser = LinkParser(story_text)
+            links = parser.extract_links()
+            for link in links:
+                response_meta = handler.status_code_handler(link)
+                writer.writerow(
+                    {
+                        "story_id": s.split(".")[0],
+                        "link": link,
+                        "response": response_meta["status_code"],
+                        "redirect_location": response_meta["redirect_location"],
+                        "redirect_response": response_meta["redirect_response"],
+                    }
+                )
+        except UnicodeDecodeError:
+            log.info(f"\nencoding error with story file {s}")
+            writer.writerow(
+                {
+                    "story_id": s.split(".")[0],
+                    "link": link,
+                    "response": "read error",
+                    "redirect_location": "",
+                    "redirect_response": "",
+                }
+            )
+        cur += 1
+        print_progress(cur, len(stories), "stories parsed")