import json
import logging
import os
import random
import re
import string
import urllib
import xml.etree.ElementTree as ET
import zipfile
from datetime import date, datetime
from datetime import timezone as datetime_timezone
from http.client import InvalidURL
from io import StringIO
from typing import List
from urllib.error import HTTPError
from uuid import uuid4
import bleach
import markdown.serializers
import yaml
from django.core.files.images import ImageFile
from markdown import Markdown
from markdown.extensions import Extension
from markdown.inlinepatterns import InlineProcessor
from wagtail.contrib.redirects.models import Redirect
from wagtail.images.models import Image
from wagtail.models import Page
from wagtail.models.media import Collection
from willow.image import UnrecognisedImageFormatError
from yaml.scanner import ScannerError
logger = logging.getLogger(__name__)
# from django.utils.dateparse import parse_date # pro hledani krome title i podle data
image_params = {} # filled on JekyllArticleImporter init and used globally
POSTS_DIR = "_posts"
# ------------------------------- Misc helper functions -------------------------------
def clone_repo(url: str) -> (str, str):
"""
Naclonuje repo do tmp s využitím gitu a vrátí cestu k němu.
Pokud URL končí lomítkem, odebereme ho, a vezmeme jako název repozitáře
string za posledním lomítkem jako název repa. To použijeme i pro promazání
takového adresáře, pokud už existuje.
"""
path = "/tmp/"
if url.endswith("/"):
url = url[:-1]
repo_name = url.split("/")[-1]
repo_path = os.path.join(path, repo_name)
os.chdir(path)
if os.path.exists(repo_path):
os.chdir(repo_path)
os.system("git pull --depth 1")
return repo_path, repo_name
os.system("git clone --depth 1 {}".format(url))
return repo_path, repo_name
def download_repo_as_zip(url: str) -> (str, str):
"""
Stáhne .zip repa, extrahuje a vrátí cestu k extrahovanému repu.
Hodně nešikovné je, že extrahovaná složka má ještě suffix "-gh-pages"
a to nevím, jestli platí vždy... regex taky pro název repa také není optimální,
ale ve finále nehraje moc roli, pokud vrátí cokoliv použitelného pro file name.
"""
path = "/tmp/"
repo_name = re.search("pirati-web/(.*)/archive/", url).group(1)
zip_path = "{}{}.zip".format(path, repo_name)
if os.path.exists(zip_path):
os.remove(zip_path)
urllib.request.urlretrieve(url, zip_path)
with zipfile.ZipFile(zip_path, "r") as zip_ref:
zip_ref.extractall(path)
# zdá se, že někdy je -gh-pages, někdy -master...
gh_pages_path = os.path.join(path, "{}-gh-pages".format(repo_name))
if os.path.exists(gh_pages_path):
return gh_pages_path, repo_name
master_path = os.path.join(path, "{}-master".format(repo_name))
if os.path.exists(master_path):
return master_path, repo_name
main_path = os.path.join(path, "{}-main".format(repo_name))
if os.path.exists(main_path):
return main_path, repo_name
raise NotImplementedError("Tento zip nedokážeme zpracovat.")
def get_or_create_image(
path: str, file_path: str, collection, repo_name: str
) -> Image or None:
"""
Funkce, která se snaží najít a vrátit Wagtail Image.
Nejdříve hledá v existujících podle cesty, resp. title...
Pak zkusí najít soubor fyzicky na disku...
Pak zkusí ještě assets/img adresář...
Pak zkusí ještě assets/img/posts adresář...
Pak zkusí stáhnout image z https://a.pirati.cz...
Pak zkusí přidat do https://a.pirati.cz "posts"...
Pak se na to vykašle...
Šla by určitě rozsekat a začistit, ale vzhledem k tomu, že je to
one-timer, tak jsme to pouze dotáhli do stavu, kdy to schroupne co nejvíce
případů.
"""
file_path = file_path.lstrip("/")
if Image.objects.filter(title=file_path).exists():
return Image.objects.filter(title=file_path).first(), ""
try:
file = ImageFile(open(os.path.join(path, file_path), "rb"), name=file_path)
image = Image(title=file_path, file=file, collection=collection)
if not image_params["dry_run"]:
image.save()
return image, ""
except (FileNotFoundError, UnrecognisedImageFormatError):
pass # cesta pomocí file_path neexisuje, jdeme dál
try:
file = ImageFile(
open(os.path.join(path, "assets/img", file_path), "rb"),
name=file_path,
)
image = Image(title=file_path, file=file, collection=collection)
if not image_params["dry_run"]:
image.save()
return image, ""
except (FileNotFoundError, UnrecognisedImageFormatError):
pass # cesta s vložením "assets/img" před file_path neexisuje, jdeme dál
try:
file = ImageFile(
open(os.path.join(path, "assets/img/posts", file_path), "rb"),
name=file_path,
)
image = Image(title=file_path, file=file, collection=collection)
if not image_params["dry_run"]:
image.save()
return image, ""
except (FileNotFoundError, UnrecognisedImageFormatError):
pass
# na disku jsme nenašli, zkusíme stáhnout z webu
fallback_name = (
"".join(random.choice(string.ascii_lowercase) for _ in range(10)) + ".jpg"
) # někdy je název obrzau spojený s poznámkou apod., připravíme si fallback name
img_name = file_path.split("/")[-1] or fallback_name
img_path = os.path.join(path, img_name)
# zkusíme, jestli obrázek není hardcoded (posty ve zlinském kraji)
if file_path.startswith("https://"):
try:
urllib.request.urlretrieve(file_path, img_path)
except (HTTPError, UnicodeEncodeError, InvalidURL, IsADirectoryError):
msg = "Nedohledán obrázek při importu článků"
log_message = "{}: URL {}\n".format(msg, file_path)
logger.warning(
log_message,
extra={"img_name": img_name, "img_url": file_path},
)
return None, log_message
# v opačném případě jdeme zkusit assets server a.pirati.cz
else:
img_assets_folder = repo_name.split(".")[0] # např. "praha" z praha.pirati.cz
img_url = "https://a.pirati.cz/resize/4000x-/{}/img/{}".format(
img_assets_folder, file_path.split("#")[0] # cistime nazev od poznamek apod
)
try:
urllib.request.urlretrieve(img_url, img_path)
except (HTTPError, UnicodeEncodeError, InvalidURL, IsADirectoryError):
try:
# druhý pokus s "posts" přidáno do URL (obvykle je ve file_path)
img_url = "https://a.pirati.cz/resize/4000x-/{}/img/posts/{}".format(
img_assets_folder, img_name.split()[0]
)
urllib.request.urlretrieve(img_url, img_path)
except (HTTPError, UnicodeEncodeError, InvalidURL, IsADirectoryError):
msg = "Nedohledán obrázek při importu článků - ani na disku, ani na URL"
log_message = "{}: cesta {}, URL {}\n".format(msg, file_path, img_url)
logger.warning(
log_message,
extra={
"file_path": file_path,
"img_name": img_name,
"img_url": img_url,
},
)
return None, log_message
file = ImageFile(open(img_path, "rb"), name=img_path)
try:
image = Image(title=file_path, file=file, collection=collection)
except UnrecognisedImageFormatError:
msg = "Obrázek byl nalezen, ale jeho formát nerozpoznán"
log_message = "{}: cesta {}\n".format(msg, file_path)
logger.warning(
log_message,
extra={
"file_path": file_path,
},
)
return None, log_message
if not image_params["dry_run"]:
try:
image.save()
except Exception as e:
msg = "Nelze uložit obrázek"
logger.warning(msg, extra={"exc": e})
return None, msg
return image, ""
def get_path_and_repo_name(url: str, use_git: bool) -> (str, str):
"""
Vrací cestu a název repozitáře podle toho zíksané různými způsoby,
podle toho jestli se jedná o odkaz na zip nebo na git.
"""
if use_git:
return clone_repo(url)
else:
return download_repo_as_zip(url)
def get_site_config(path) -> dict:
"""
Vrací config Jekyll repa jako dict.
"""
with open(os.path.join(path, "_config.yml")) as f:
config = yaml.safe_load(f.read())
return config
def get_title_from_site_config(site_config: dict) -> str:
if "title" in site_config:
return " - " + site_config.get("title", "")
return ""
def unmark_element(element, stream=None):
"""
Očišťuje element (perex) od ostatních značek
"""
if stream is None:
stream = StringIO()
if element.text:
stream.write(element.text)
for sub in element:
unmark_element(sub, stream)
if element.tail:
stream.write(element.tail)
return stream.getvalue()
# ------------------- Setup markdown extensions and settings -----------------------
class ImgProcessor(InlineProcessor):
def handleMatch(self, m, data):
el = ET.Element("embed")
el.attrib["embedtype"] = "image"
el.attrib["alt"] = m.group(1)
el.attrib["format"] = "left"
parsed_image_path = JekyllArticleImporter.get_parsed_file_path(m.group(2))
image_obj, _ = get_or_create_image(
path=image_params["path"],
file_path=parsed_image_path,
collection=image_params["collection"],
repo_name=image_params["repo_name"],
)
if not image_obj:
return None, m.start(0), m.end(0)
el.attrib["id"] = str(image_obj.pk)
return el, m.start(0), m.end(0)
class ImgExtension(Extension):
def extendMarkdown(self, md):
IMG_PATTERN = r"!\[(.*?)\]\((.*?)\)"
md.inlinePatterns.register(ImgProcessor(IMG_PATTERN, md), "img", 175)
# Wagtail to portrebuje
# https://docs.wagtail.io/en/stable/extending/rich_text_internals.html#data-format
markdown.serializers.HTML_EMPTY.add("embed")
Markdown.output_formats["plain"] = unmark_element
plain_md = Markdown(output_format="plain")
plain_md.stripTopLevelTags = False
html_md = Markdown(extensions=[ImgExtension()])
params = {}
# ------------------------------- Importer class -------------------------------
class JekyllArticleImporter:
def __init__(
self,
article_parent_page_id: int,
article_parent_page_model,
collection_id: int,
url: str,
dry_run: bool,
use_git: bool,
page_model,
):
self.page_model = page_model
# Params
self.article_parent_page_id = article_parent_page_id
self.article_parent_page_model = article_parent_page_model
self.collection = Collection.objects.get(id=collection_id)
self.dry_run = dry_run
self.use_git = use_git
self.url = url
# Computed proprs
import time
time.sleep(5)
self.article_parent_page = self.article_parent_page_model.objects.filter(
id=self.article_parent_page_id
).first()
self.path, self.repo_name = get_path_and_repo_name(self.url, self.use_git)
self.site = self.article_parent_page.get_site()
self.site_config = get_site_config(self.path)
self.article_path = self.site_config.get("articlepath", None)
self.permalink = self.site_config.get("permalink", None)
self.title_suffix = get_title_from_site_config(self.site_config)
# Counters
self.success_counter = 0
self.exists_counter = 0
self.skipped_counter = 0
self.page_log = "" # output saved on page instance
# Filling global var for ImgParser
image_params["path"] = self.path
image_params["collection"] = self.collection
image_params["repo_name"] = self.repo_name
image_params["dry_run"] = self.dry_run
def create_redirects(self, article, match):
y = match.group(1)
m = match.group(2)
d = match.group(3)
slug = match.group(4)
if self.article_path: # asi jenom Ceske Budejovice
Redirect.objects.get_or_create(
site=self.site,
old_path="/%s/%s/%s/%s/%s"
% (self.article_path, y, m.zfill(2), d.zfill(2), slug),
defaults={"is_permanent": True, "redirect_page": article},
)
elif self.permalink:
Redirect.objects.get_or_create(
site=self.site,
old_path=self.permalink.replace(":title", slug),
defaults={"is_permanent": True, "redirect_page": article},
)
def create_summary_log(self):
"""
Podle (aktuálních) hodnot counterů přidá do self.page_log
různé zprávy pro uživatele.
"""
self.page_log += "==================================\n"
if self.success_counter:
base_msg = "Úspěšně otestováno" if self.dry_run else "Úspěšně naimportováno"
self.page_log += "{} {} článků\n".format(base_msg, self.success_counter)
if self.exists_counter:
self.page_log += "z toho {} již existovalo\n".format(self.exists_counter)
if self.skipped_counter:
self.page_log += "NELZE importovat {} článků\n".format(self.skipped_counter)
self.article_parent_page.last_import_log = self.page_log
self.article_parent_page.save_revision()
@staticmethod
def get_parsed_file_path(path: str):
"""
Získá cestu z proměnné v "{{ }}" závorkách
"""
if "{{" in path:
try:
parsed_path = path.split("{{")[1].split("|")[0].split("'")[1]
except IndexError:
parsed_path = path.split("{{")[1].split("|")[0].split('"')[1]
return parsed_path
else:
return path
@staticmethod
def get_perex(text):
text = re.split(r"^\s*$", text.strip(), flags=re.MULTILINE)[0]
return plain_md.convert(text)
def handle_content(self, article, meta: dict, html: str):
"""
Převádí naparsované html do stremfieldů.
Pokud meta info článku obsahuje "fancybox" - tzn. v článku je galerie,
tak nejdříve očistíme HTML od for loopů galerie a podtom
v "self.handle_fancybox_gallery" získáme JSON data pro GalleryBlock
z ArticleMixinu.
Pokud článek galerii nemá (drtivá většina), tak uložíme jako
RichText (block type text).
"""
if meta.get("fancybox", None):
# Galerie josu v HTML ve formě dvou "{% for" tagů,
# ty potřebujeme zahodit z texxtu
html = re.sub(
"{% for(.*?){% for(.*?){% endfor %}(.*?){% endfor %}",
"",
html,
flags=re.DOTALL,
)
if "{% for" in html: # pro případ, že by byl jenom jeden
html = re.sub("{% for(.*?){% endfor %}", "", html, flags=re.DOTALL)
text_data_dict = {"type": "text", "value": html}
gallery_data_dict = self.handle_fancybox_gallery(article, meta)
article.content = json.dumps([text_data_dict, gallery_data_dict])
else:
text_data_dict = {"type": "text", "value": html}
article.content = json.dumps([text_data_dict])
def handle_fancybox_gallery(self, article, meta: dict) -> dict:
for gallery in meta["fancybox"]:
# gallery by měl být dict s name a img
gallery_name = gallery.get("name", "")
gallery_images = gallery.get("img", []) or []
if not len(gallery_images):
self.page_log += (
"{} Nepodařilo se získat obrázky v galerii {}\n".format(
article.title, gallery_name
)
)
continue
wagtail_image_list = []
for img in gallery_images:
if not img.get("src"):
self.page_log += (
'{}: Obrázek {} v galerii namá atribut "src" \n'.format(
article.title, img.get("title", "")
)
)
continue
wagtail_image, log_message = get_or_create_image(
self.path, img["src"], self.collection, self.repo_name
)
wagtail_image_list.append(wagtail_image)
if log_message:
self.page_log += "{}: {}".format(article.title, log_message)
data = {
"type": "gallery",
"value": {"gallery_items": []},
"id": str(uuid4()),
}
if not wagtail_image_list:
return data
for image in wagtail_image_list:
data["value"]["gallery_items"].append(
{"type": "item", "value": image.id, "id": str(uuid4())}
)
return data
@staticmethod
def handle_meta_is_str(meta: str) -> dict:
"""
Snaží se vyřešit situaci, že meta se nenaparsovala na dict, ale na string,
kde je sice klíč:hodnota, ale další klíč následuje za mezerou po předchozí
hodnotě.
Iteruju teda přes rozesekaný string přes dvojtečky, ale každá value kromě
poslední (viz if idx == len(string_parts) - 2) má za poslední mezerou klíč
pro další hodnotu. Takže položku z další iterace splitnu přes mezery,
spojím všechny kromě poslední do value a tu přiřadím klíči z aktuální iterace.
Poslední položka iterace už je samotná value, takže tam handluju jinak.
Protože sahám na +1, tak hlídám číslo iterace pro přeskočení poslední (kde
už je položka iterace pouze value bez key, takže jí beru
v před-předposlední...).
"""
logger.info(
"Meta se neparsuje na dict, ale na str - zkouším složitější parse",
extra={"article_meta": meta},
)
string_parts = meta.split(":")
meta_dict = {}
for idx, part in enumerate(string_parts):
if idx == len(string_parts) - 1:
break
key = part.split()[-1]
if idx == len(string_parts) - 2:
value = string_parts[idx + 1]
else:
value = " ".join(string_parts[idx + 1].split()[0:-1])
meta_dict.update({key: value})
return meta_dict
def handle_tags(self, article, meta):
article.tags.clear()
tags = meta.get("tags", []) or [] # někdy jsou tags None
if type(tags) == str: # někdy jsou tags str
if "\n" in tags:
tags = tags.split("\n")
elif " " in tags:
tags = tags.split()
else:
tags = [tags]
for tag_name in tags:
try:
article.tags.add(tag_name)
except ValueError:
try:
article.tags.add(str(tag_name))
except Exception:
msg = "Nelze importovat tag"
logger.warning(msg, extra={"tag": tag_name})
self.page_log += "{} - {}\n".format(msg, tag_name)
article.save_revision()
def import_post(self, file_path):
with open(os.path.join(self.path, file_path), "rt") as f:
r = re.split(r"^---\s*$", f.read(), maxsplit=3, flags=re.MULTILINE)
try:
meta = yaml.safe_load(r[1].replace("\t", ""))
except (ScannerError, ValueError, IndexError):
msg = "Nelze importovat článek - neparsovatelný YAML"
logger.warning(msg, extra={"file_path": file_path})
self.page_log += "{} - {}\n".format(msg, file_path)
self.skipped_counter += 1
return None
if isinstance(meta, str): # pokud se špatně naparsovalo meta (není dict)
meta = self.handle_meta_is_str(meta)
try:
title = meta["title"]
except TypeError:
msg = "Nelze importovat článek - nepodařilo se získat title"
logger.warning(msg, extra={"article_meta": meta})
self.page_log += "{} - {}\n".format(msg, meta)
self.skipped_counter += 1
return None
try:
article = (
self.article_parent_page.get_descendants().get(title=title).specific
)
self.exists_counter += 1
except (Page.DoesNotExist, Page.MultipleObjectsReturned):
article = self.page_model()
md = r[2] # "raw" markdown z postu
html = html_md.convert(md)
# očistíme o případné nechtěné HTML tagy
html = bleach.clean(
html, tags=list(bleach.sanitizer.ALLOWED_TAGS) + ["div", "p"]
)
article.perex = self.get_perex(md) or "..."
self.handle_content(article, meta, html)
if meta.get("date", None):
meta_date = meta["date"]
if isinstance(meta_date, date):
article.timestamp = datetime(
meta_date.year, meta_date.month, meta_date.day
).replace(tzinfo=datetime_timezone.utc)
else:
parsed_date = meta["date"].split()[0]
if parsed_date:
article.timestamp = datetime.strptime(
parsed_date[0:10], "%Y-%m-%d"
).replace(tzinfo=datetime_timezone.utc)
else:
log_message = (
"Článek {} má nesprávné datum: {}, nastavuji dnešní".format(
title, meta["date"]
)
)
logger.warning(log_message)
self.page_log += "{} - {}\n".format(log_message, meta)
article.timestamp = timestamp.now()
else:
filename = os.path.basename(file_path)
parsed_date = "-".join(filename.split("-")[:3])
if parsed_date:
article.timestamp = datetime.strptime(parsed_date, "%Y-%m-%d").replace(
tzinfo=datetime_timezone.utc
)
else:
log_message = (
"Článek {} má nesprávné datum: {}, nastavuji dnešní".format(
title, meta["date"]
)
)
logger.warning(log_message)
self.page_log += "{} - {}\n".format(log_message, meta)
article.title = meta["title"]
article.author = meta.get("author", "Česká pirátská strana")
article.seo_title = article.title + self.title_suffix
article.search_description = meta.get("description", "")
if meta.get("image", None):
article.image, log_message = get_or_create_image(
self.path, meta["image"], self.collection, self.repo_name
)
if log_message:
self.page_log += "{}: {}".format(article.title, log_message)
if self.dry_run:
return article
try:
if not article.id:
self.article_parent_page.add_child(instance=article)
logger.info("Vytvářím článek: %s" % article)
rev = article.save_revision()
if meta.get("published", True):
rev.publish()
except Exception as e:
msg = "Nelze uložit importovaný článek"
logger.warning(
msg,
extra={"article_title": article.title, "exception": e},
)
self.page_log += "{} - {} - {}\n".format(msg, article.title, e)
self.skipped_counter += 1
return article
self.handle_tags(article, meta)
self.success_counter += 1
return article
def perform_import(self) -> "List[dict]":
"""
Projde adresář článků a pokusí se zprocesovat Markdown do article.
Vrací list dict pro django messages (klíč levelu, text).
Začne vyčištěním logu.
"""
try:
self.article_parent_page.last_import_log = ""
self.article_parent_page.save()
msg = "{} Import započat".format(datetime.now())
logger.info(msg)
self.page_log += "{}\n\n".format(msg)
for file_name in os.listdir(os.path.join(self.path, POSTS_DIR)):
# Případ podsložek (typicky po jednotlivých letech)
if os.path.isdir(os.path.join(self.path, POSTS_DIR, file_name)):
posts_sub_folder = os.path.join(self.path, POSTS_DIR, file_name)
for sub_file_name in os.listdir(posts_sub_folder):
file_path = os.path.join(posts_sub_folder, sub_file_name)
self.process_article(sub_file_name, file_path)
# Případ všech článků v jedné složce
else:
file_path = os.path.join(POSTS_DIR, file_name)
self.process_article(file_name, file_path)
msg = "{} Import ukončen".format(datetime.now())
logger.info(msg)
self.page_log += "{}\n\n".format(msg)
self.create_summary_log()
finally:
import_lock_filename = f"/tmp/.{self.article_parent_page_id}.import-lock"
if os.path.exists(import_lock_filename):
os.remove(import_lock_filename)
def process_article(self, file_name: str, file_path: str):
match = re.match(r"(\d*)-(\d*)-(\d*)-(.*)\.(.*)", file_name)
if match:
ext = match.group(5)
if ext == "md":
article = self.import_post(file_path)
if self.dry_run or not article:
return
try:
article.save() # ujistím se, že mám "redirect_page" pro Redirect uloženou
self.create_redirects(article, match)
except Exception as e:
msg = "{}: nelze uložit - {}".format(article.title, e)
logger.error(msg)
self.page_log += "{}\n".format(msg)
self.skipped_counter += 1
else:
msg = "Nepodporovaná přípona souboru: %s" % ext
logger.warning(msg)
self.page_log += "{}\n".format(msg)
self.skipped_counter += 1
else:
msg = "Přeskočeno: %s" % file_name
logger.warning(msg)
self.page_log += "{}\n".format(msg)
self.skipped_counter += 1