263 lines
8.1 KiB
Python
263 lines
8.1 KiB
Python
from typing import Optional, List
|
|
from sqlmodel import Field, Relationship, SQLModel
|
|
|
|
from arcology.parse import parse_sexp, print_sexp
|
|
|
|
class Link(SQLModel, table=True):
|
|
__tablename__ = "arcology_links"
|
|
source_title: Optional[str] = Field(default="", description="The title of the page the link is written in.")
|
|
|
|
def get_source_title(self):
|
|
return parse_sexp(self.source_title)
|
|
|
|
source_id: str = Field(primary_key=True, foreign_key="arcology_nodes.node_id")
|
|
source_node: Optional["Node"] = Relationship(
|
|
sa_relationship_kwargs=dict(
|
|
# back_populates="outlinks",
|
|
primaryjoin="Node.node_id == Link.source_id"
|
|
)
|
|
)
|
|
|
|
dest_id: str = Field(primary_key=True, foreign_key="arcology_nodes.node_id")
|
|
dest_node: Optional["Node"] = Relationship(
|
|
sa_relationship_kwargs=dict(
|
|
# back_populates="backlinks",
|
|
primaryjoin="Node.node_id == Link.dest_id"
|
|
)
|
|
)
|
|
|
|
source_file: str = Field(primary_key=True, foreign_key="arcology_pages.file")
|
|
source_page: Optional["Page"] = Relationship(
|
|
back_populates="outlinks",
|
|
sa_relationship_kwargs=dict(
|
|
primaryjoin="Page.file==Link.source_file"
|
|
)
|
|
)
|
|
|
|
dest_file: str = Field(primary_key=True, foreign_key="arcology_pages.file")
|
|
dest_page: Optional["Page"] = Relationship(
|
|
back_populates="backlinks",
|
|
sa_relationship_kwargs=dict(
|
|
primaryjoin="Page.file==Link.dest_file"
|
|
)
|
|
)
|
|
from sqlmodel import Session, select
|
|
import hashlib
|
|
|
|
from arcology.key import ArcologyKey, id_to_arcology_key
|
|
import arcology.html as html
|
|
|
|
class Page(SQLModel, table=True):
|
|
__tablename__ = "arcology_pages"
|
|
file: str = Field(primary_key=True)
|
|
key: str = Field(description="The ARCOLOGY_KEY for the page")
|
|
title: str = Field(description="Primary title of the page")
|
|
hash: str = Field(description="The hash of the file when it was indexed")
|
|
root_id: str = Field(description="The ID for the page itself", foreign_key="nodes.node_id")
|
|
site: str = Field(description="Maps to an arcology.Site key.")
|
|
allow_crawl: str = Field(description="Lisp boolean for whether this page should go in robots.txt")
|
|
|
|
nodes: List["Node"] = Relationship(
|
|
back_populates="page",
|
|
sa_relationship_kwargs=dict(
|
|
primaryjoin="Node.file==Page.file"
|
|
)
|
|
)
|
|
tags: List["Tag"] = Relationship(
|
|
sa_relationship_kwargs=dict(
|
|
primaryjoin="Tag.file==Page.file"
|
|
)
|
|
)
|
|
references: List["Reference"] = Relationship(
|
|
sa_relationship_kwargs=dict(
|
|
primaryjoin="Reference.file==Page.file"
|
|
)
|
|
)
|
|
|
|
def get_title(self):
|
|
return parse_sexp(self.title)
|
|
|
|
def get_key(self):
|
|
return parse_sexp(self.key)
|
|
|
|
def get_file(self):
|
|
return parse_sexp(self.file)
|
|
|
|
def get_arcology_key(self):
|
|
return ArcologyKey(self.get_key())
|
|
|
|
def get_site(self):
|
|
return self.get_arcology_key().site
|
|
|
|
backlinks: List["Link"] = Relationship(
|
|
back_populates="dest_page",
|
|
sa_relationship_kwargs=dict(
|
|
primaryjoin="Page.file==Link.dest_file"
|
|
)
|
|
)
|
|
|
|
outlinks: List["Link"] = Relationship(
|
|
back_populates="source_page",
|
|
sa_relationship_kwargs=dict(
|
|
primaryjoin="Page.file==Link.source_file"
|
|
)
|
|
)
|
|
|
|
backlink_pages: List["Page"] = Relationship(
|
|
link_model=Link,
|
|
back_populates="outlink_pages",
|
|
sa_relationship_kwargs=dict(
|
|
foreign_keys="[Link.dest_file]",
|
|
viewonly=True,
|
|
)
|
|
)
|
|
|
|
outlink_pages: List["Page"] = Relationship(
|
|
link_model=Link,
|
|
back_populates="backlink_pages",
|
|
sa_relationship_kwargs=dict(
|
|
foreign_keys="[Link.source_file]",
|
|
viewonly=True,
|
|
)
|
|
)
|
|
@classmethod
|
|
def from_file(cls, path: str, session: Session):
|
|
q = select(cls).where(cls.file==print_sexp(path))
|
|
return session.exec(q).one()
|
|
|
|
@classmethod
|
|
def from_key(cls, key: str, session: Session):
|
|
q = select(cls).where(cls.key==print_sexp(key))
|
|
try:
|
|
return next(session.exec(q))
|
|
except StopIteration:
|
|
return None
|
|
def make_backlinks_org(self):
|
|
if self.backlinks is None:
|
|
return ''
|
|
|
|
def to_org(link: Link):
|
|
return \
|
|
"""
|
|
* [[id:{path}][{title}]]
|
|
""".format(
|
|
path=parse_sexp(link.source_id),
|
|
title=link.get_source_title()
|
|
)
|
|
|
|
return '\n'.join([ to_org(link) for link in self.backlinks ])
|
|
|
|
async def document_html(self):
|
|
cache_key = parse_sexp(self.hash)
|
|
return html.gen_html(parse_sexp(self.file), cache_key)
|
|
|
|
async def backlink_html(self):
|
|
org = self.make_backlinks_org()
|
|
cache_key = hashlib.sha224(org.encode('utf-8')).hexdigest()
|
|
return html.gen_html_text(org, cache_key)
|
|
class Tag(SQLModel, table=True):
|
|
__tablename__ = "arcology_tags"
|
|
file: str = Field(primary_key=True, foreign_key="arcology_pages.file")
|
|
tag: str = Field(primary_key=True, description="The tag itself.")
|
|
node_id: str = Field(description="A heading ID which the tag applies to")
|
|
|
|
def tag(self):
|
|
return parse_sexp(self.tag)
|
|
class Node(SQLModel, table=True):
|
|
__tablename__ = "arcology_nodes"
|
|
node_id: str = Field(primary_key=True, description="The heading ID property")
|
|
file: str = Field(description="File in which this Node appears", foreign_key="arcology_pages.file")
|
|
level: str = Field(description="Outline depth of the heading. 0 is top-level")
|
|
|
|
page: Optional["Page"] = Relationship(
|
|
back_populates="nodes",
|
|
sa_relationship_kwargs=dict(
|
|
viewonly=True,
|
|
primaryjoin="Node.file==Page.file"
|
|
)
|
|
)
|
|
class Reference(SQLModel, table=True):
|
|
__tablename__ = "arcology_refs"
|
|
file: str = Field(primary_key=True, foreign_key="arcology_pages.file")
|
|
ref: str = Field(primary_key=True, description="The full URI of the reference itself.")
|
|
node_id: str = Field(description="A heading ID which the ref applies to")
|
|
|
|
def url(self):
|
|
return parse_sexp(self.ref)
|
|
class Keyword(SQLModel, table=True):
|
|
__tablename__ = "keywords"
|
|
file: str = Field(primary_key=True, foreign_key="arcology_pages.file")
|
|
keyword: str = Field(primary_key=True, description="")
|
|
value: str = Field(description="The value of the page")
|
|
|
|
def filename(self):
|
|
return parse_sexp(self.file)
|
|
|
|
def keyword(self):
|
|
return parse_sexp(self.keyword)
|
|
|
|
def value(self):
|
|
return parse_sexp(self.value)
|
|
|
|
@classmethod
|
|
def get(cls, key: str, value: str, session: Session):
|
|
q = select(cls).where(cls.keyword==print_sexp(key)).where(cls.value==print_sexp(value))
|
|
try:
|
|
return next(session.exec(q))
|
|
except StopIteration:
|
|
return None
|
|
class Feed(SQLModel, table=True):
|
|
__tablename__ = "arcology_feeds"
|
|
file: str = Field(primary_key=True, foreign_key="arcology_pages.file")
|
|
key: str = Field(primary_key=True, description="The routing key for the feed.")
|
|
title: str = Field(description="Title of the page which the feed is embedded in")
|
|
site: str = Field(description="Arcology Site which the feed resides on.")
|
|
post_visibility: str = Field(description="Visibility of the feed's posts in feed2toot, etc")
|
|
|
|
def get_key(self):
|
|
return parse_sexp(self.key)
|
|
|
|
def get_arcology_key(self):
|
|
return ArcologyKey(self.get_key())
|
|
|
|
def get_title(self):
|
|
return parse_sexp(self.title)
|
|
|
|
def get_site(self):
|
|
return parse_sexp(self.site)
|
|
|
|
def get_post_visibility(self):
|
|
return parse_sexp(self.post_visibility)
|
|
|
|
def dict(self, **kwargs):
|
|
return dict(
|
|
key=self.get_key(),
|
|
url=self.get_arcology_key().to_url(),
|
|
title=self.get_title(),
|
|
site=self.get_site(),
|
|
visibility=self.get_post_visibility(),
|
|
)
|
|
|
|
from sqlmodel import create_engine
|
|
from sqlalchemy import event
|
|
|
|
from arcology.config import get_settings
|
|
|
|
from pathlib import Path
|
|
|
|
settings = get_settings()
|
|
org_roam_sqlite_file_name = Path(settings.org_roam_db).expanduser().resolve()
|
|
arroyo_sqlite_file_name = Path(settings.arcology_db).expanduser().resolve()
|
|
|
|
def make_engine():
|
|
engine = create_engine('sqlite:///{path}'.format(path=arroyo_sqlite_file_name), echo=False)
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def do_connect(dbapi_connection, _connection_record):
|
|
dbapi_connection.execute("attach database '{orgdb}' as orgroam;".format(orgdb=org_roam_sqlite_file_name))
|
|
|
|
return engine
|
|
|
|
|
|
engine = make_engine()
|