Refresh the format registry gathering code and actions

.github/workflows/data-update.yml

@@ -34,6 +34,9 @@ jobs:
       - name: Update from various data sources...
         run: ./update.sh
 
+      - name: Generate derivatives...
+        run: ./derive.sh
+
       - name: Deploy updated site...
         run: ./deploy.sh
         env:

.gitignore

@@ -8,3 +8,8 @@
 /bin
 /pywikibot.lwp
 /passwordfile
+*.pyc
+/registries.db
+/.venv
+/build
+/data

Makefile

@@ -0,0 +1,9 @@
+
+all: registries.db
+
+registries.db: foreging/*.py digipres.github.io/_sources/registries/*
+	rm -fr data
+	mkdir -p data
+	python -m foreging.populate --json data
+	cp data/registries.db digipres.github.io/_data/formats/registries.db
+	cp data/*.parquet digipres.github.io/_data/formats/index/

README.md

@@ -35,6 +35,8 @@ To Do
 * http://en.wikipedia.org/wiki/Alphabetical_list_of_filename_extensions_%28M%E2%80%93R%29
 * http://www.webarchive.org.uk/aadda-discovery/formats?f[0]=content_type_ext:%22.bmp%22
 * https://twitter.com/benfinoradin/status/532212803630039041
+* Talk about how to use `git submodule update --recursive --remote` to make sure `pywikibot` and `digipres.github.io` are up to date.
+* Using `uvx datasette serve data/registries.db` to quickly poke around in the database.
 
 COPTR Bot
 ---------

aggregates.py

@@ -86,7 +86,6 @@ def addFormat(rid,fid,finfo):
     # And add:
     fmts[rid]['formats'][fid] = finfo
 
-
 def aggregateFDD():
     rid = "fdd"
     print("Parsing %s..." % rid)
@@ -132,6 +131,7 @@ def aggregateFDD():
                     if rid in fmts: # FIXME this needs to be more robust, rather than relying on happening after 'addFormat' is called for the first time.
                         fmts[rid]['warnings'].append(f"Error when parsing XML from '{filename}': {e}")
 
+
 def aggregateTRiD():
     rid = "trid"
     print("Parsing %s..." % rid)
@@ -394,9 +394,9 @@ def aggregateWikiData():
 with open("%s/extensions.yml" % data_dir, 'w') as outfile:
     outfile.write( yaml.safe_dump(extensions, default_flow_style=False) )
 
-# Write out Venn data
+# Write out Venn data, starting from a list like [extension] -> Registry_ID:
 print("Outputting Venn data based on extensions...")
-# Key all the RID-to-integer mappings:
+# Key all the Registry_ID-to-integer mappings:
 vennls = {}
 i = 0
 for fmt in fmts:
@@ -407,15 +407,19 @@ def aggregateWikiData():
 venndsl = defaultdict(list)
 vennlt = defaultdict(int)
 vennids = {}
+# Loop over all extensions:
 for extension in exts:
     regs = set()
     regIds = set()
+    # Loop over each registry the extension appears in:
     for ridder in exts[extension]['identifiers']:
         regs.add(vennls[ridder['regId']])
         regIds.add(ridder['regId'])
     for rid in regs:
         vennlt[rid] += 1
+    # Build a unique key for each registry combination:
     key = ','.join(sorted(regs))
+    # Use the key to build up each overlap set:
     vennids[key] = sorted(regIds)
     venndsl[key].append(extension)
     vennds[key] += 1

derive.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+set -e
+
+source venv/bin/activate
+
+make
+
+cp data/registries.db digipres.github.io/_data/formats/
+cp data/*.parquet digipres.github.io/_data/formats/index
+

foreging/__init__.py

@@ -0,0 +1 @@
+# FOrmat REGistry INdexinG

foreging/coptr.py

@@ -0,0 +1,72 @@
+import mwclient as mw
+from mwclient.listing import Category, PageList
+import mwparserfromhell
+
+from .models import Software
+
+import logging
+logging.basicConfig(level=logging.WARNING)
+
+
+coptr_host = 'coptr.digipres.org'
+user_agent = 'DigiPresFormatIndexClient/0.1 (andrew.jackson@dpconline.org)'
+site = mw.Site(coptr_host, path='/', clients_useragent=user_agent)
+
+#for tool_page in site.allpages():
+#    pass
+
+#category = site.categories[u"Tool Grid"]
+#for page in category:
+#    print(page.name)
+
+
+# {{Infobox tool
+# |image=JHOVE.gif
+# |purpose=JHOVE provides functions to perform format-specific identification, validation, and characterization of digital objects.
+# |homepage=http://jhove.openpreservation.org/
+# |license=GNU Lesser General Public License (LGPL)
+# |platforms=JHOVE should be usable on any UNIX, Windows, or OS X platform with an appropriate J2SE installation. It should run on any operating system that supports Java 1.5 and has a directory-based file system.
+# |formats_in=EPUB, GIF, JP2, JPEG, PDF, PNG, PREMIS (Preservation Metadata Implementation Strategies), TIFF, WARC, XML, AIFF, WAVE, GZIP, ASCII, UTF-8, HTML, MP3
+# |function=Encryption Detection, File Format Identification, Metadata Extraction, Validation
+# }}
+
+# FIXME this does both at once! One should write the page info needed to JSON. The other should use it.
+# But, we don't know everything we need yet, I guess?
+
+category: PageList = site.categories[u"Tools"]
+for page in category:
+    print(page.name)
+    text = page.text()
+    wikicode = mwparserfromhell.parse(text)
+    templates = wikicode.filter_templates(matches='infobox tool')
+    template = templates[0]
+    formats = template.get("formats_in", None)
+    if formats:
+        formats = [f.strip() for f in formats.value.split(",")]
+        print(f"  <  {formats}")
+    formats = template.get("formats_out", None)
+    if formats:
+        formats = [f.strip() for f in formats.value.split(",")]
+        print(f"  >  {formats}")
+    print(page.pageid)
+    if isinstance(page, Category):
+        for member in page.members():
+            print(f"{page.name} > {member.name}")
+    else:
+        pass
+        s = Software(
+            id=f"coptr:pageid:{page.pageid}",
+            name=page.name,
+            version=None,
+            license=None,
+            registry_url=f"https://{coptr_host}/Special:Redirect/page/{page.pageid}"
+        )
+        license = template.get('license', None)
+        if license:
+            s.license = license.value.strip()
+        print(s)
+
+
+
+# Workflows in Workflow namespace
+# Formats is another potential category, but needs patching in via external IDs.

foreging/db/extension_sets.py

@@ -0,0 +1,43 @@
+import json
+import sqlite3
+import logging
+import argparse
+from collections import defaultdict
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def generate_ext_sets(db):
+    con = sqlite3.connect(db)
+
+    cur = con.cursor()
+
+    ext_sets = defaultdict(set)
+    ext_counts = defaultdict(int)
+    for row in cur.execute("SELECT registry_id, format.id, e.value FROM format, json_each(extensions) AS e ORDER BY e.value ASC"):
+        ext_sets[row[0]].add(row[2].lower().strip())
+        ext_counts[row[0]] += 1
+
+    for source, ext_set in ext_sets.items():
+        ext_sets[source] = list(ext_set)
+        logger.info(f"Registry {source} has {ext_counts[source]} extensions, of which {len(ext_set)} are unique. Ratio: {ext_counts[source]/len(ext_set)}")
+    return ext_sets, ext_counts
+
+
+if __name__ == "__main__":
+    # Args setup:
+    parser = argparse.ArgumentParser()
+    parser.add_argument('input_db')
+    parser.add_argument('output_json')
+    args = parser.parse_args()
+
+    # Query and return the sets of extensions:
+    ext_sets, ext_counts = generate_ext_sets(args.input_db)
+
+    # Output the sets of extensions:
+    with open(args.output_json, 'w') as f: 
+        json.dump(ext_sets, f)
+
+
+

foreging/db/models.py

@@ -0,0 +1,117 @@
+from datetime import date
+from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, JSON, Column
+
+
+class Registry(SQLModel, table=True):
+    id: str | None = Field(default=None, primary_key=True)
+    name: str = Field(index=True)
+    url: str | None = Field()
+    id_prefix: str | None = Field()
+    index_data_url: str | None = Field()
+
+    data_log: list["RegistryDataLogEntry"] = Relationship()
+
+
+class RegistryDataLogEntry(SQLModel, table=True):
+    __tablename__ = 'registry_data_log'
+    id: int | None = Field(default=None, primary_key=True)
+    level: str = Field(index=True)
+    message: str = Field()
+    url: str | None = Field()
+
+    registry_id: str | None = Field(default=None, foreign_key="registry.id")
+    registry: Registry | None = Relationship(back_populates="data_log")
+
+    # Define how to spot unique entries in a set
+    def __hash__(self):
+        return hash(self.message)
+    def __eq__(self,other):
+        return self.message == other.message
+
+class SoftwareReadsFormatLink(SQLModel, table=True):
+    __tablename__ = "formats_read_by_software"
+    format_id: str | None = Field(default=None, foreign_key="format.id", primary_key=True)
+    software_id: str | None = Field(default=None, foreign_key="software.id", primary_key=True)
+
+class SoftwareWritesFormatLink(SQLModel, table=True):
+    __tablename__ = "formats_written_by_software"
+    format_id: str | None = Field(default=None, foreign_key="format.id", primary_key=True)
+    software_id: str | None = Field(default=None, foreign_key="software.id", primary_key=True)
+
+class Software(SQLModel, table=True):
+    id: str | None = Field(default=None, primary_key=True)
+    name: str = Field(index=True)
+    version: str | None = Field(index=True)
+    summary: str | None = Field(index=True)
+    license: str | None = Field(index=True)
+    registry_url: str | None = Field(index=True)
+
+    reads: list["Format"] = Relationship(back_populates="readers", link_model=SoftwareReadsFormatLink)
+    writes: list["Format"] = Relationship(back_populates="writers", link_model=SoftwareWritesFormatLink)
+
+    registry_id: str | None = Field(default=None, foreign_key="registry.id")
+    registry: Registry | None = Relationship()
+
+    # Define how to spot unique entries in a set
+    def __hash__(self):
+        return hash(self.id)
+    def __eq__(self,other):
+        return self.id == other.id
+
+class FormatGenresLink(SQLModel, table=True):
+    __tablename__ = "format_genres"
+    format_id: str | None = Field(default=None, foreign_key="format.id", primary_key=True)
+    genre_id: str | None = Field(default=None, foreign_key="genre.id", primary_key=True)
+
+class Genre(SQLModel, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    name: str = Field(index=True)
+    #
+    formats: list["Format"] = Relationship(back_populates="genres", link_model=FormatGenresLink)
+
+    # Define how to spot unique entries in a set
+    def __hash__(self):
+        return hash(self.name)
+    def __eq__(self,other):
+        return self.name == other.name
+
+class MediaTypesFormatsLink(SQLModel, table=True):
+    __tablename__ = "format_media_types"
+    format_id: str | None = Field(default=None, foreign_key="format.id", primary_key=True)
+    media_type_id: str | None = Field(default=None, foreign_key="media_type.id", primary_key=True)
+
+class MediaType(SQLModel, table=True):
+    __tablename__ = "media_type"
+    id: str | None = Field(default=None, primary_key=True)
+    #
+    formats: list["Format"] = Relationship(back_populates="media_types", link_model=MediaTypesFormatsLink)
+
+    # Define how to spot unique entries in a set
+    def __hash__(self):
+        return hash(self.id)
+    def __eq__(self,other):
+        return self.id == other.id
+
+class Format(SQLModel, table=True):
+    id: str | None = Field(default=None, primary_key=True)
+    name: str | None = Field(index=True)
+    version: str | None = Field(index=True)
+    summary: str | None = Field(index=True)
+    genres: list["Genre"] = Relationship(back_populates="formats", link_model=FormatGenresLink)
+    extensions: list[str] | None = Field(default=None, sa_column=Column(JSON))
+    media_types: list["MediaType"] = Relationship(back_populates="formats", link_model=MediaTypesFormatsLink)
+    has_magic: bool = Field(default=False)
+    primary_media_type: str | None = Field(index=True)
+    parent_media_type: str | None = Field(index=True)
+    registry_url: str | None = Field(index=True)
+    registry_source_data_url: str | None = Field(index=True)
+    registry_index_data_url: str | None = Field(index=True)
+    created: date | None = Field(index=True)
+    last_modified: date | None = Field(index=True)
+
+    readers: list["Software"] = Relationship(back_populates="reads", link_model=SoftwareReadsFormatLink)
+    writers: list["Software"] = Relationship(back_populates="writes", link_model=SoftwareWritesFormatLink)
+
+    registry_id: str | None = Field(default=None, foreign_key="registry.id")
+    registry: Registry | None = Relationship()
+