path: root/gitsrht/blueprints/repo.py

import binascii
import os
import pygit2
import pygments
import sys
import subprocess
from datetime import timedelta
from jinja2 import Markup
from flask import Blueprint, render_template, abort, send_file, request
from flask import Response, url_for
from gitsrht.editorconfig import EditorConfig
from gitsrht.git import Repository as GitRepository, commit_time, annotate_tree
from gitsrht.git import diffstat
from gitsrht.rss import generate_feed
from io import BytesIO
from pygments import highlight
from pygments.lexers import guess_lexer, guess_lexer_for_filename, TextLexer
from pygments.formatters import HtmlFormatter
from scmsrht.access import get_repo, get_repo_or_redir
from scmsrht.formatting import get_formatted_readme, get_highlighted_file
from scmsrht.redis import redis
from scmsrht.urls import get_clone_urls
from srht.config import cfg
from srht.markdown import markdown

repo = Blueprint('repo', __name__)

def get_readme(repo, tip, link_prefix=None):
    if not tip:
        return None

    def file_finder(name):
        try:
            blob = tip.tree[name]
        except KeyError:
            return None, None

        if blob and blob.type == "blob":
            return blob.id.hex, blob
        return None, None

    def content_getter(blob):
        return repo.get(blob.id).data.decode()

    return get_formatted_readme("git.sr.ht:git", file_finder, content_getter,
            link_prefix=link_prefix)

def _highlight_file(name, data, blob_id):
    return get_highlighted_file("git.sr.ht:git", name, blob_id, data)

def render_empty_repo(owner, repo):
    origin = cfg("git.sr.ht", "origin")
    urls = get_clone_urls(origin, owner, repo, 'git@{origin}:{user}/{repo}')
    return render_template("empty-repo.html", owner=owner, repo=repo,
            clone_urls=urls)

def get_last_3_commits(commit):
    commits = [commit]
    for parent in commit.parents:
        commits.append(parent)
        for grandparent in parent.parents:
            commits.append(grandparent)

    commits = sorted(commits, key=lambda c: commit_time(c), reverse=True)
    return commits[:3]

@repo.route("/<owner>/<repo>")
def summary(owner, repo):
    owner, repo = get_repo_or_redir(owner, repo)

    with GitRepository(repo.path) as git_repo:
        if git_repo.is_empty:
            return render_empty_repo(owner, repo)

        default_branch = git_repo.default_branch()
        tip = git_repo.get(default_branch.target)
        commits = get_last_3_commits(tip)
        link_prefix = url_for(
            'repo.tree', owner=repo.owner, repo=repo.name,
            ref=default_branch.name)
        readme = get_readme(git_repo, tip,
            link_prefix=link_prefix)
        tags = [(ref, git_repo.get(git_repo.references[ref].target))
            for ref in git_repo.listall_references()
            if ref.startswith("refs/tags/")]
        tags = [tag for tag in tags
                if isinstance(tag[1], pygit2.Tag) or isinstance(tag[1], pygit2.Commit)]
        tags = sorted(tags, key=lambda c: commit_time(c[1]), reverse=True)
        latest_tag = tags[0] if len(tags) else None
        return render_template("summary.html", view="summary",
                owner=owner, repo=repo, readme=readme, commits=commits,
                latest_tag=latest_tag, default_branch=default_branch,
                is_annotated=lambda t: isinstance(t, pygit2.Tag))

def lookup_ref(git_repo, ref, path):
    ref = ref or git_repo.default_branch().name[len("refs/heads/"):]
    if path is None:
        path = []
    else:
        path = path.split("/")
    commit = None
    try:
        commit = git_repo.revparse_single(ref)
    except KeyError:
        pass
    except ValueError:
        pass
    while commit is None and len(path):
        ref += "/" + path[0]
        path = path[1:]
        try:
            commit = git_repo.revparse_single(ref)
        except KeyError:
            pass
        except ValueError:
            pass
    if commit is None:
        abort(404)
    if isinstance(commit, pygit2.Tag):
        commit = git_repo.get(commit.target)
    return commit, ref, "/".join(path)

@repo.route("/<owner>/<repo>/tree", defaults={"ref": None, "path": ""})
@repo.route("/<owner>/<repo>/tree/<path:ref>", defaults={"path": ""})
@repo.route("/<owner>/<repo>/tree/<ref>/<path:path>")
def tree(owner, repo, ref, path):
    owner, repo = get_repo_or_redir(owner, repo)
    with GitRepository(repo.path) as git_repo:
        if git_repo.is_empty:
            return render_empty_repo(owner, repo)

        commit, ref, path = lookup_ref(git_repo, ref, path)

        tree = commit.tree
        if not tree:
            abort(404)
        editorconfig = EditorConfig(git_repo, tree, path)

        path = path.split("/")
        for part in path:
            if part == "":
                continue
            if not tree or part not in tree:
                abort(404)
            entry = tree[part]
            if entry.type == "blob":
                tree = annotate_tree(git_repo, tree, commit)
                commit = next(
                        (e.commit for e in tree if e.name == entry.name), None)
                blob = git_repo.get(entry.id)
                data = None
                if not blob.is_binary:
                    try:
                        data = blob.data.decode()
                    except:
                        data = '[unable to decode]'
                return render_template("blob.html", view="tree",
                        owner=owner, repo=repo, ref=ref, path=path, entry=entry,
                        blob=blob, data=data, commit=commit,
                        highlight_file=_highlight_file,
                        editorconfig=editorconfig)
            tree = git_repo.get(entry.id)

        if not tree:
            abort(404)
        tree = annotate_tree(git_repo, tree, commit)
        tree = sorted(tree, key=lambda e: e.name)

        return render_template("tree.html", view="tree", owner=owner, repo=repo,
                ref=ref, commit=commit, tree=tree, path=path)

@repo.route("/<owner>/<repo>/blob/<path:ref>/<path:path>")
def raw_blob(owner, repo, ref, path):
    owner, repo = get_repo_or_redir(owner, repo)
    with GitRepository(repo.path) as git_repo:
        commit, ref, path = lookup_ref(git_repo, ref, path)

        blob = None
        entry = None
        tree = commit.tree
        path = path.split("/")
        for part in path:
            if part == "":
                continue
            if part not in tree:
                abort(404)
            entry = tree[part]
            if entry.type == "blob":
                tree = annotate_tree(git_repo, tree, commit)
                commit = next(e.commit for e in tree if e.name == entry.name)
                blob = git_repo.get(entry.id)
                break
            tree = git_repo.get(entry.id)

        if not blob:
            abort(404)

        return send_file(BytesIO(blob.data),
                as_attachment=blob.is_binary,
                attachment_filename=entry.name,
                mimetype="text/plain" if not blob.is_binary else None)

@repo.route("/<owner>/<repo>/archive/<path:ref>.tar.gz")
def archive(owner, repo, ref):
    owner, repo = get_repo_or_redir(owner, repo)
    with GitRepository(repo.path) as git_repo:
        commit, ref, _ = lookup_ref(git_repo, ref, None)

        path = f"/tmp/{commit.id.hex}{binascii.hexlify(os.urandom(8))}.tar.gz"
        try:
            args = [
                "git",
                "--git-dir", repo.path,
                "archive",
                "--format=tar.gz",
                "--prefix", f"{repo.name}-{ref}/",
                "-o", path, ref
            ]
            subp = subprocess.run(args, timeout=30,
                    stdout=sys.stdout, stderr=sys.stderr)
        except:
            try:
                os.unlink(path)
            except:
                pass
            raise

        if subp.returncode != 0:
            try:
                os