"""
Gitweb action handlers (git_*): summary, tree, blob, log, commit, tag, etc.
Ported from gitweb/gitweb.perl action handlers. Invoked by main.dispatch.
"""

from __future__ import annotations

import base64
import mimetypes
import os
from datetime import UTC, datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Literal, TypedDict
from urllib.parse import parse_qsl, quote, urlencode

import pygit2
from fastapi import HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse, RedirectResponse, Response
from python_ripgrep import PySortMode, PySortModeKind
from python_ripgrep import search as rg_search

from pygitweb.auth import can_read_branch
from pygitweb.config import BLOB_LANG, settings
from pygitweb.formatting import age_string, sanitize, to_utf8
from pygitweb.git_helpers import (
	get_blob_at_ref_path,
	get_blob_unified_diff,
	get_commit_history,
	get_commit_unified_diff,
	get_commits_in_range,
	get_readme_at_ref_path,
	get_tree_at_ref_path,
	git_get_heads_list,
	git_get_project_description,
	git_get_references,
	git_get_remotes_info,
	git_get_tags_list,
	git_get_type,
	open_repo,
	parse_commit,
	parse_tag,
)
from pygitweb.hooks_install import list_installable_hooks, list_installed_hooks
from pygitweb.merge_requests import merge_request_tag_response, resolve_merge_request_tag_to_tip
from pygitweb.projects import git_get_project_owner
from pygitweb.templates_env import (
	Cell,
	env,
	html_format,
	html_join,
	html_link,
	render_page,
	trusted_html,
)
from pygitweb.validation import is_valid_pathname, is_valid_ref_format


class BlameHunkInfo(TypedDict):
	commit_id: str
	commit_msg: str
	commit_author: str
	time: str
	line_start: int
	line_end: int
	author_name: str
	time_display: str


def _split_query_list(values: list[str]) -> list[str]:
	parts: list[str] = []
	for v in values:
		for seg in (v or "").split(","):
			s = seg.strip()
			if s:
				parts.append(s)
	return parts


def _parse_query_bool(value: str | None) -> bool | None:
	if value is None:
		return None
	v = value.strip().lower()
	if v in ("1", "true", "yes", "on"):
		return True
	if v in ("0", "false", "no", "off"):
		return False
	raise HTTPException(status_code=400, detail=f"Invalid boolean: {value!r}")


def _project_worktree_root(project: str) -> Path:
	root = (Path(settings.PROJECTROOT) / project).resolve()
	if not root.exists() or not root.is_dir():
		raise HTTPException(status_code=404, detail="Project directory not found")
	return root


def _validated_search_paths(project_root: Path, raw_paths: list[str]) -> list[str]:
	root = project_root.resolve()
	if not raw_paths:
		return [str(root)]

	paths: list[str] = []
	for raw in raw_paths:
		p = (raw or "").strip()
		if not p:
			continue
		candidate = Path(p)
		if candidate.is_absolute() or ":" in p:
			raise HTTPException(status_code=400, detail="path not relative to the project root")
		abs_path = (root / p).resolve()
		if not abs_path.is_relative_to(root):
			raise HTTPException(status_code=400, detail="path not within the project root")
		if not abs_path.exists():
			raise HTTPException(status_code=400, detail=f"path entry does not exist: {p}")
		paths.append(str(abs_path))

	if not paths:
		return [str(root)]
	return paths


_SORT_KINDS: tuple[str, ...] = ("Path", "LastModified", "LastAccessed", "Created")

SearchSortQuery = Literal[
	"Path",
	"-Path",
	"LastModified",
	"-LastModified",
	"LastAccessed",
	"-LastAccessed",
	"Created",
	"-Created",
]

SearchFlagQuery = Literal["true", "false"]


def _parse_search_sort(value: str | None) -> PySortMode | None:
	if value is None or not value.strip():
		return None
	raw = value.strip()
	reverse = False
	if raw.startswith("-"):
		reverse = True
		raw = raw[1:].strip()
	if raw not in _SORT_KINDS:
		allowed = ", ".join(_SORT_KINDS)
		raise HTTPException(status_code=400, detail=f"Invalid sort kind; allowed: {allowed}")
	kind = getattr(PySortModeKind, raw)
	return PySortMode(kind=kind, reverse=reverse)


def _pagination_url(request: Request, page: int, pagecount: int) -> str:
	"""Build URL for a pagination page, preserving path and other query params."""
	params = dict(parse_qsl(request.url.query, keep_blank_values=True))
	params["page"] = str(page)
	params["pagecount"] = str(pagecount)
	return f"{request.url.path}?{urlencode(params)}"


def parse_pagination(
	page: str | None,
	pagecount: str | None,
) -> tuple[int, int]:
	"""Parse page and pagecount; default 1 and 25; raise if pagecount > 50."""
	p = 1
	pc = 25
	if page is not None:
		try:
			p = int(page)
		except ValueError as e:
			raise HTTPException(status_code=400, detail="page must be an integer") from e
		if p < 1:
			raise HTTPException(status_code=400, detail="page must be at least 1")
	if pagecount is not None:
		try:
			pc = int(pagecount)
		except ValueError as e:
			raise HTTPException(status_code=400, detail="pagecount must be an integer") from e
		if pc < 1:
			raise HTTPException(status_code=400, detail="pagecount must be at least 1")
		if pc > 50:
			raise HTTPException(status_code=400, detail="pagecount must not exceed 50")
	return p, pc


def _tree_url(project: str, h: str, f: str | None, a: str = "tree") -> str:
	"""Build URL for tree or blob: /project/{project}?a=a&h=h&f=f with f quoted."""
	q = f"a={a}&h={quote(h, safe='')}"
	if f:
		q += f"&f={quote(f, safe='/')}"
	return f"/project/{project}?{q}"


def _format_date(epoch: int | None, tz_offset: int | None = None) -> str:
	"""Format epoch timestamp to readable date string.
	tz_offset is in minutes (as returned by pygit2).
	"""
	if epoch is None:
		return ""
	try:
		tz = timezone(timedelta(seconds=tz_offset * 60)) if tz_offset is not None else UTC
		dt = datetime.fromtimestamp(epoch, tz=tz)
		return dt.strftime("%Y-%m-%d %H:%M:%S")
	except (ValueError, OSError):
		return ""


def _format_commit_table_row(project: str, commit: dict[str, Any], short: bool = False) -> list[Cell]:
	"""Format a single commit as a table row."""
	oid = commit.get("oid", "")
	oid_short = oid[:7] if oid else ""
	subject = commit.get("subject", "")
	author = commit.get("author", "")
	author_epoch = commit.get("author_epoch")

	date_str = _format_date(author_epoch, commit.get("author_tz"))
	age_sec = None
	if author_epoch:
		age_sec = datetime.now(UTC).timestamp() - author_epoch
		age_str = age_string(age_sec) if age_sec > 0 else "right now"
	else:
		age_str = ""

	commit_href = f"/project/{project}?a=commit&h={quote(oid, safe='')}"
	diff_href = f"/project/{project}?a=commitdiff&h={quote(oid, safe='')}"
	author_display = author or "unknown"
	commit_cell = html_link(commit_href, oid_short)
	diff_cell = html_link(diff_href, "diff")

	if short:
		return [commit_cell, subject, author_display, age_str, diff_cell]
	return [commit_cell, subject, author_display, date_str, age_str, diff_cell]


def _pagination_context(
	request: Request,
	page: int,
	pagecount: int,
	has_prev: bool,
	has_next: bool,
	total_pages: int | None = None,
) -> dict[str, Any]:
	"""Context for pagination.html."""
	prev_url = _pagination_url(request, page - 1, pagecount) if has_prev else ""
	next_url = _pagination_url(request, page + 1, pagecount) if has_next else ""
	page_links = None
	if total_pages is not None and total_pages <= 20:
		page_links = [(p, _pagination_url(request, p, pagecount)) for p in range(1, total_pages + 1)]
	elif total_pages is not None:
		start = max(1, page - 2)
		end = min(total_pages, page + 2)
		page_links = [(p, _pagination_url(request, p, pagecount)) for p in range(start, end + 1)]
		if start > 1:
			page_links = [(1, _pagination_url(request, 1, pagecount))] + [(-1, "")] + page_links
		if end < total_pages:
			page_links = page_links + [(-1, "")] + [(total_pages, _pagination_url(request, total_pages, pagecount))]
	return {
		"current_page": page,
		"pagecount": pagecount,
		"has_prev": has_prev,
		"has_next": has_next,
		"prev_url": prev_url,
		"next_url": next_url,
		"total_pages": total_pages,
		"page_links": page_links,
	}


def _render_tree_table(project: str, ref: str | None, f: str | None) -> str:
	"""Return an HTML table listing files and directories at ref/path."""
	if f is not None and not is_valid_pathname(f):
		return ""
	result = get_tree_at_ref_path(project, ref, f)
	if not result:
		return '<p class="text-muted mb-0">No files found.</p>'
	tree, ref_oid = result
	entries = [(obj.name, obj.type_str, str(obj.id)) for obj in tree]
	dirs = sorted((n, t, o) for n, t, o in entries if t == "tree")
	blobs = sorted((n, t, o) for n, t, o in entries if t == "blob")
	rows: list[list[Cell]] = []
	for name, _, _ in dirs:
		if name is None:
			continue
		sub_path = f"{f.strip('/')}/{name}".strip("/") if f else name
		href = _tree_url(project, ref_oid, sub_path)
		rows.append([html_link(href, f"{name}/"), "tree", ""])
	for name, _, _ in blobs:
		if name is None:
			continue
		sub_path = f"{f.strip('/')}/{name}".strip("/") if f else name
		href = _tree_url(project, ref_oid, sub_path, a="blob")
		blame_href = _tree_url(project, ref_oid, sub_path, a="blame")
		rows.append([
			html_link(href, name),
			"blob",
			html_link(blame_href, "blame"),
		])
	if not rows:
		return '<p class="text-muted mb-0">Empty directory.</p>'
	return env.get_template("table.html").render(cols=["Name", "Type", ""], rows=rows)


def _table_page(
	*,
	title: str,
	heading: str | None = None,
	cols: list[str] | None = None,
	rows: list[list[Cell]] | None = None,
	project: str | None = None,
	pagination: dict[str, Any] | None = None,
	empty_message: str | None = None,
	empty_muted: bool = False,
) -> HTMLResponse:
	ctx: dict[str, object] = {
		"title": title,
		"heading": title if heading is None else heading,
		"project": project,
		"pagination": pagination,
		"empty_message": empty_message,
		"empty_muted": empty_muted,
	}
	if cols is not None:
		ctx["cols"] = cols
		ctx["rows"] = rows or []
	return render_page("table_page.html", **ctx)


def _render_readme_card(
	project: str,
	ref_oid: str,
	readme_filename: str,
	readme_content: str,
	blob_dir: str = "",
) -> str:
	"""Return HTML for the README section (and script for markdown).

	blob_dir is the current tree path for relative links.
	"""
	is_markdown = readme_filename.lower().endswith(".md")
	if is_markdown:
		return env.get_template("readme_card.html").render(
			markdown=True,
			readme_b64=base64.b64encode(readme_content.encode("utf-8")).decode("ascii"),
			blob_base=f"/project/{project}?a=blob&h={quote(ref_oid, safe='')}&f=",
			blob_dir=blob_dir,
		)
	return env.get_template("readme_card.html").render(
		markdown=False,
		plain=sanitize(readme_content) or "",
	)


def _commit_unified_diff_or_raise(project: str, h: str) -> tuple[str, str]:
	"""Return (unified_diff_text, oid_short) for commit h. Raises HTTPException on error."""
	if not h:
		raise HTTPException(status_code=400, detail="Commit hash (h) required")
	if not is_valid_ref_format(h):
		raise HTTPException(status_code=400, detail="Invalid ref or hash")
	try:
		return get_commit_unified_diff(project, h)
	except (KeyError, pygit2.GitError, OSError, ValueError) as e:
		raise HTTPException(status_code=404, detail="Commit not found") from e


class SummaryRefOption(TypedDict):
	label: str
	value: str
	kind: str


class SummaryRefsPayload(TypedDict):
	options: list[SummaryRefOption]


class SummaryRefState(TypedDict):
	ref: str
	display: str
	commit_href: str
	browse_href: str
	log_href: str
	shortlog_href: str
	tree_html: str
	readme_html: str


def summary_ref_options(project: str) -> list[SummaryRefOption]:
	options: list[SummaryRefOption] = []
	for name, ref, _oid in git_get_heads_list(project):
		options.append({"label": name, "value": ref, "kind": "branch"})
	for name, ref, _oid in git_get_tags_list(project):
		options.append({"label": name, "value": ref, "kind": "tag"})
	return options


def summary_ref_state(project: str, requested_ref: str | None) -> SummaryRefState:
	ref = requested_ref or "HEAD"
	commit_oid = ""
	browse_ref = ref
	try:
		repo = open_repo(project)
		obj = repo.revparse_single(ref)
	except (KeyError, pygit2.GitError, OSError):
		ref = "HEAD"
		try:
			repo = open_repo(project)
			obj = repo.revparse_single(ref)
		except (KeyError, pygit2.GitError, OSError):
			obj = None
	if isinstance(obj, pygit2.Commit):
		commit_oid = str(obj.id)
	elif isinstance(obj, pygit2.Tag):
		try:
			target_obj = repo[obj.target]
		except (KeyError, pygit2.GitError):
			target_obj = None
		if isinstance(target_obj, pygit2.Commit):
			commit_oid = str(target_obj.id)
		elif target_obj is not None:
			browse_ref = str(target_obj.id)
	tree_html = _render_tree_table(project, ref, None)
	readme_html = ""
	readme = get_readme_at_ref_path(project, ref, "")
	if readme:
		readme_filename, readme_content = readme
		readme_html = _render_readme_card(project, ref, readme_filename, readme_content, "")
	log_ref = commit_oid or ref
	return {
		"ref": ref,
		"display": (commit_oid[:7] if commit_oid else ref) or "N/A",
		"commit_href": f"/project/{project}?a=commit&h={quote(commit_oid or ref, safe='')}",
		"browse_href": f"/project/{project}?a=tree&h={quote(browse_ref, safe='')}",
		"log_href": f"/project/{project}?a=log&h={quote(log_ref, safe='')}",
		"shortlog_href": f"/project/{project}?a=shortlog&h={quote(log_ref, safe='')}",
		"tree_html": tree_html,
		"readme_html": readme_html,
	}


# ---------- Action handlers ----------


def git_object(project: str, h: str | None) -> Response:
	"""Show object by type: commit, tree, tag, or blob. Dispatches to the appropriate view."""
	if not h:
		raise HTTPException(status_code=400, detail="Object hash (h) required")
	if not is_valid_ref_format(h):
		raise HTTPException(status_code=400, detail="Invalid ref or hash")
	obj_type = git_get_type(project, h)
	if not obj_type:
		raise HTTPException(status_code=404, detail="Object not found")
	if obj_type == "commit":
		return git_commit(project, h)
	if obj_type == "tree":
		return git_tree(project, h, None)
	if obj_type == "tag":
		return git_tag(project, h)
	if obj_type == "blob":
		try:
			repo = open_repo(project)
			obj = repo.revparse_single(h)
		except (KeyError, pygit2.GitError, OSError) as e:
			raise HTTPException(status_code=404, detail="Object not found") from e
		if not isinstance(obj, pygit2.Blob):
			raise HTTPException(status_code=404, detail="Not a blob")
		data = obj.data
		text = to_utf8(data) or ""
		lines = text.split("\n")
		num_lines = max(1, len(lines))
		oid_short = str(obj.id)[:7]
		return render_page(
			"blob.html",
			title=f"Blob {oid_short} - {project}",
			heading=f"Blob {oid_short}",
			project=project,
			code=sanitize(text) or "",
			line_nums="\n".join(str(i) for i in range(1, num_lines + 1)),
			lang="",
			blame=False,
		)
	raise HTTPException(status_code=404, detail="Unknown object type")


def git_search(
	project: str,
	search_patterns: str | None,
	search_paths: str | None,
	search_globs: str | None,
	search_heading: SearchFlagQuery | None,
	search_multiline: SearchFlagQuery | None,
	search_sort: SearchSortQuery | None,
	search_max_count: str | None,
) -> JSONResponse:
	patterns = _split_query_list([search_patterns] if search_patterns is not None else [])
	if not patterns:
		raise HTTPException(status_code=400, detail="patterns is required (repeat ?patterns= or use comma-separated)")

	project_root = _project_worktree_root(project)
	paths = _validated_search_paths(project_root, _split_query_list([search_paths] if search_paths is not None else []))

	globs = _split_query_list([search_globs] if search_globs is not None else [])
	heading = _parse_query_bool(search_heading)
	multiline = _parse_query_bool(search_multiline)
	sort = _parse_search_sort(search_sort)

	max_count: int = 25
	if search_max_count is not None and search_max_count.strip() != "":
		try:
			max_count = int(search_max_count)
		except ValueError as e:
			raise HTTPException(status_code=400, detail="max_count must be an integer") from e
		if max_count < 1:
			raise HTTPException(status_code=400, detail="max_count must be at least 1")

	try:
		results = rg_search(
			patterns=patterns,
			paths=paths,
			globs=(globs or None),
			heading=heading,
			sort=sort,
			max_count=max_count,
			multiline=multiline,
		)
	except Exception as e:
		raise HTTPException(status_code=500, detail=f"Search failed: {e}") from e

	root_str = str(project_root.resolve())
	rel_paths: list[str] = []
	for p in paths:
		abs_p = str(Path(p).resolve())
		rel = os.path.relpath(abs_p, root_str)
		rel_paths.append("" if rel == "." else rel.replace(os.sep, "/"))

	return JSONResponse({
		"project": project,
		"project_root": root_str,
		"patterns": patterns,
		"paths": rel_paths,
		"globs": globs,
		"heading": heading,
		"sort": (search_sort or None),
		"max_count": max_count,
		"multiline": multiline,
		"results": results,
	})


def git_search_page(project: str) -> HTMLResponse:
	"""Per-project search UI page. Calls /project/{project}?a=search via fetch."""
	project_enc = quote(project, safe="/")
	return render_page(
		"search.html",
		title=f"Search - {project}",
		project=project,
		project_enc=project_enc,
		project_url=f"/project/{project_enc}",
		search_api_url=f"/project/{project_enc}",
	)


def _hooks_context(project: str, project_enc: str) -> dict[str, Any]:
	"""Data for hooks_summary_cell.html."""
	try:
		installed = list_installed_hooks(project)
		available = list_installable_hooks(project)
	except (KeyError, OSError):
		return {"mode": "unavailable", "project_enc": project_enc, "installed": [], "available": []}
	return {
		"mode": "manage",
		"project_enc": project_enc,
		"installed": installed,
		"available": available,
	}


def _merge_request_context(project: str) -> dict[str, Any] | None:
	heads = git_get_heads_list(project)
	if not heads:
		return None
	refs = [h[1] for h in heads]
	branches = [{"short": h[0], "ref": h[1]} for h in heads]
	if "refs/heads/main" in refs:
		ours_default_ref = "refs/heads/main"
	elif "refs/heads/master" in refs:
		ours_default_ref = "refs/heads/master"
	else:
		ours_default_ref = refs[0]
	theirs_default_ref = next((r for r in refs if r != ours_default_ref), refs[0])
	return {
		"project": project,
		"branches": branches,
		"theirs_default_ref": theirs_default_ref,
		"ours_default_ref": ours_default_ref,
	}


def _project_has_boards(project: str) -> bool:
	try:
		return len(git_get_references(project, "refs/tags/boards")) > 0
	except Exception:
		return False


BLOG_MANIFEST_PATH: str = "articles/manifest.json"


def project_has_blog(project: str) -> bool:
	try:
		return get_blob_at_ref_path(project, "HEAD", BLOG_MANIFEST_PATH) is not None
	except Exception:
		return False


class BoardLink(TypedDict):
	href: str
	text: str
	style: str | None


class BlogLink(TypedDict):
	href: str
	text: str
	style: str | None


def board_link_spec(project: str) -> BoardLink:
	if _project_has_boards(project):
		return {
			"href": f"/project/{quote(project, safe='/')}/board/",
			"text": "project board",
			"style": None,
		}
	style = "color: #999; cursor: not-allowed;" if settings.AUTH else None
	return {
		"href": "/board/create?project=" + quote(project, safe=""),
		"text": "create board",
		"style": style,
	}


def blog_link_spec(project: str) -> BlogLink:
	if project_has_blog(project):
		return {
			"href": f"/project/{quote(project, safe='/')}/blog/",
			"text": "view blog",
			"style": None,
		}
	return {
		"href": "#",
		"text": "no blog",
		"style": "color: #999; cursor: not-allowed;",
	}


def create_board_link(project: str) -> Cell:
	href = "/board/create?project=" + quote(project, safe="")
	if settings.AUTH:
		return html_link(href, "create board", style="color: #999; cursor: not-allowed;")
	return html_link(href, "create board")


def project_board_link(project: str) -> Cell:
	spec = board_link_spec(project)
	if spec["style"]:
		return html_link(spec["href"], spec["text"], style=spec["style"])
	return html_link(spec["href"], spec["text"])


def git_summary(project: str, extra_rows: list[list[Cell]] | None = None) -> HTMLResponse:
	"""Project summary page. Port of git_summary."""
	descr = git_get_project_description(project) or ""
	owner = git_get_project_owner(project) or ""
	initial_ref_state = summary_ref_state(project, "HEAD")
	project_enc = quote(project, safe="/")
	ref_view: dict[str, object] = dict(initial_ref_state)
	ref_view["tree_html"] = trusted_html(initial_ref_state["tree_html"])
	ref_view["readme_html"] = trusted_html(initial_ref_state["readme_html"])
	return render_page(
		"summary.html",
		title=f"{settings.SITE_NAME} - {project}",
		project=project,
		descr=descr,
		owner=owner,
		ref=ref_view,
		project_url=f"/project/{project}",
		project_url_enc=f"/project/{project_enc}",
		board=board_link_spec(project),
		blog=blog_link_spec(project),
		hooks=_hooks_context(project, project_enc),
		merge_request=_merge_request_context(project),
		extra_rows=extra_rows or [],
	)


def git_remotes(project: str) -> HTMLResponse:
	"""Remotes page: list configured remotes (name, url, push_url)."""
	remotes = git_get_remotes_info(project)
	title = f"Remotes - {project}"
	if not remotes:
		return _table_page(title=title, heading="Remotes", empty_message="No remotes configured.")
	rows = [[r["name"], r["url"] or "—", r["push_url"] or "—"] for r in remotes]
	return _table_page(title=title, heading="Remotes", cols=["Name", "URL", "Push URL"], rows=rows)


def git_blob(project: str, h: str | None, f: str | None, *, raw: bool = False) -> Response:
	"""Return blob content when f points to a file. Raw = bytes; else HTML with escaped content."""
	if not f:
		raise HTTPException(status_code=400, detail="File path (f) required")
	if not is_valid_pathname(f):
		raise HTTPException(status_code=400, detail="Invalid path")
	result = get_blob_at_ref_path(project, h, f)
	if not result:
		raise HTTPException(status_code=404, detail="File not found")
	blob, _ = result
	data = blob.data
	if raw:
		media_type, _ = mimetypes.guess_type(f.split("/")[-1])
		if media_type is None:
			try:
				data.decode("utf-8")
				media_type = "text/plain; charset=utf-8"
			except UnicodeDecodeError:
				media_type = "application/octet-stream"
		return Response(content=data, media_type=media_type)
	text = to_utf8(data) or ""
	lines = text.split("\n")
	num_lines = max(1, len(lines))
	ext = os.path.splitext(f.split("/")[-1])[1].lstrip(".").lower()
	return render_page(
		"blob.html",
		title=f"{f} - {project}",
		code=sanitize(text) or "",
		line_nums="\n".join(str(i) for i in range(1, num_lines + 1)),
		lang=BLOB_LANG.get(ext, ""),
		blame=False,
	)


def git_blog_raw(project: str, path: str) -> Response:
	"""Serve a blob from HEAD for the project blog viewer."""
	if not is_valid_pathname(path):
		raise HTTPException(status_code=404, detail="File not found")
	result = get_blob_at_ref_path(project, "HEAD", path)
	if not result:
		raise HTTPException(status_code=404, detail="File not found")
	blob, _ = result
	data = blob.data
	media_type, _ = mimetypes.guess_type(path.split("/")[-1])
	if media_type is None:
		try:
			data.decode("utf-8")
			media_type = "text/plain; charset=utf-8"
		except UnicodeDecodeError:
			media_type = "application/octet-stream"
	return Response(content=data, media_type=media_type)


def _blame_hunk_end_line(hunk: pygit2.BlameHunk) -> int:
	return hunk.final_start_line_number + hunk.lines_in_hunk - 1


def _merge_blame_hunks(repo: pygit2.Repository, blame: pygit2.Blame) -> list[BlameHunkInfo]:
	out: list[BlameHunkInfo] = []
	i = 0
	n = len(blame)
	while i < n:
		hunk = blame[i]
		cid = str(hunk.final_commit_id)
		line_start = hunk.final_start_line_number
		line_end = _blame_hunk_end_line(hunk)
		j = i + 1
		while j < n:
			nh = blame[j]
			if str(nh.final_commit_id) != cid:
				break
			if nh.final_start_line_number != line_end + 1:
				break
			line_end = _blame_hunk_end_line(nh)
			j += 1
		try:
			bc = repo[cid].peel(pygit2.Commit)
		except (KeyError, pygit2.GitError, ValueError) as e:
			raise HTTPException(status_code=500, detail=f"Missing blame commit {cid}") from e
		auth = bc.author
		tzinfo = timezone(timedelta(minutes=auth.offset))
		dt = datetime.fromtimestamp(float(auth.time), tz=tzinfo)
		out.append({
			"commit_id": cid,
			"commit_msg": bc.message.rstrip("\n"),
			"commit_author": f"{auth.name} <{auth.email}>",
			"time": dt.isoformat(),
			"line_start": line_start,
			"line_end": line_end,
			"author_name": auth.name,
			"time_display": _format_date(auth.time, auth.offset),
		})
		i = j
	return out


def _blame_for_file(project: str, h: str | None, f: str) -> list[BlameHunkInfo]:
	if not is_valid_pathname(f):
		raise HTTPException(status_code=400, detail="Invalid path")
	if not get_blob_at_ref_path(project, h, f):
		raise HTTPException(status_code=404, detail="File not found")
	repo = open_repo(project)
	base_ref = h or (str(repo.head.target) if repo.head else None)
	if not base_ref:
		raise HTTPException(status_code=404, detail="No revision to blame")
	try:
		commit_tip = repo.revparse_single(base_ref).peel(pygit2.Commit)
	except (KeyError, pygit2.GitError, ValueError) as e:
		raise HTTPException(status_code=400, detail="Invalid revision for blame") from e
	path_clean = f.strip("/")
	try:
		blame = repo.blame(path_clean, newest_commit=commit_tip.id)
	except (pygit2.GitError, OSError) as e:
		raise HTTPException(status_code=400, detail=f"Blame failed: {e}") from e
	return _merge_blame_hunks(repo, blame)


def _blame_line_links(project: str, hunks: list[BlameHunkInfo], num_lines: int) -> list[dict[str, str] | None]:
	hunk_by_start = {h["line_start"]: h for h in hunks}
	lines: list[dict[str, str] | None] = []
	for line_no in range(1, num_lines + 1):
		hunk = hunk_by_start.get(line_no)
		if hunk is None:
			lines.append(None)
			continue
		cid = hunk["commit_id"]
		lines.append({
			"href": f"/project/{project}?a=commit&h={quote(cid, safe='')}",
			"title": f"{hunk['author_name']} — {hunk['time_display']}",
			"oid_short": cid[:7],
		})
	return lines


def git_blame(project: str, h: str | None, f: str | None) -> HTMLResponse:
	"""Blame page: file content with per-range commit links and line numbers."""
	if not f:
		raise HTTPException(status_code=400, detail="File path (f) required")
	result = get_blob_at_ref_path(project, h, f)
	if not result:
		raise HTTPException(status_code=404, detail="File not found")
	blob, _ref_oid = result
	hunks = _blame_for_file(project, h, f)
	text = to_utf8(blob.data) or ""
	lines = text.split("\n")
	num_lines = max(1, len(lines))
	ext = os.path.splitext(f.split("/")[-1])[1].lstrip(".").lower()
	return render_page(
		"blob.html",
		title=f"Blame {f} - {project}",
		code=sanitize(text) or "",
		line_nums="\n".join(str(i) for i in range(1, num_lines + 1)),
		lang=BLOB_LANG.get(ext, ""),
		blame=True,
		blame_lines=_blame_line_links(project, hunks, num_lines),
	)


def git_blame_raw(project: str, h: str | None, f: str | None) -> JSONResponse:
	"""Per-file blame as JSON: merged contiguous hunks per commit with line ranges."""
	if not f:
		raise HTTPException(status_code=400, detail="File path (f) required")
	hunks = _blame_for_file(project, h, f)
	out = [
		{
			"commit_id": h["commit_id"],
			"commit_msg": h["commit_msg"],
			"commit_author": h["commit_author"],
			"time": h["time"],
			"line_range": (
				str(h["line_start"]) if h["line_start"] == h["line_end"] else f"{h['line_start']}-{h['line_end']}"
			),
		}
		for h in hunks
	]
	return JSONResponse(out)


def git_tree(project: str, h: str | None, f: str | None) -> HTMLResponse:
	"""Tree page: list files and directories; directories link to tree with f=path."""
	if f is not None and not is_valid_pathname(f):
		raise HTTPException(status_code=400, detail="Invalid path")
	result = get_tree_at_ref_path(project, h, f)
	if not result:
		raise HTTPException(status_code=404, detail="Tree or path not found")
	_tree, ref_oid = result
	crumbs: list[dict[str, str]] = []
	if f:
		parts = f.strip("/").split("/")
		for i, seg in enumerate(parts):
			prefix = "/".join(parts[: i + 1])
			crumbs.append({"href": _tree_url(project, ref_oid, prefix), "label": seg})
	title_path = f" / {f}" if f else ""
	readme_html = ""
	needs_showdown = False
	readme = get_readme_at_ref_path(project, ref_oid, f or "")
	if readme:
		readme_filename, readme_content = readme
		needs_showdown = readme_filename.lower().endswith(".md")
		readme_html = _render_readme_card(project, ref_oid, readme_filename, readme_content, f or "")
	return render_page(
		"tree.html",
		title=f"{project}{title_path} - Tree",
		project=project,
		project_href=f"/project/{project}",
		breadcrumbs=crumbs,
		path=f or "",
		tree_html=trusted_html(_render_tree_table(project, h, f)),
		readme_html=trusted_html(readme_html),
		needs_showdown=needs_showdown,
	)


def git_log(project: str, h: str | None, request: Request, page: int, pagecount: int) -> HTMLResponse:
	"""Commit log page. Port of git_log."""
	skip = (page - 1) * pagecount
	commits = get_commit_history(project, ref=h, max_count=pagecount + 1, skip=skip)
	if not commits:
		raise HTTPException(status_code=404, detail="No commits found")
	has_next = len(commits) > pagecount
	if has_next:
		commits = commits[:pagecount]
	rows = []
	for commit in commits:
		rows.append(_format_commit_table_row(project, commit, short=False))
	ref_display = h[:7] if h else "HEAD"
	title = f"Log - {project} @ {ref_display}"
	return _table_page(
		title=title,
		cols=["Commit", "Subject", "Author", "Date", "Age", "Diff"],
		rows=rows,
		pagination=_pagination_context(request, page, pagecount, page > 1, has_next),
	)


def git_shortlog(project: str, h: str | None, request: Request, page: int, pagecount: int) -> HTMLResponse:
	"""Shortlog page. Port of git_shortlog."""
	skip = (page - 1) * pagecount
	commits = get_commit_history(project, ref=h, max_count=pagecount + 1, skip=skip)
	if not commits:
		raise HTTPException(status_code=404, detail="No commits found")
	has_next = len(commits) > pagecount
	if has_next:
		commits = commits[:pagecount]
	rows = []
	for commit in commits:
		rows.append(_format_commit_table_row(project, commit, short=True))
	ref_display = h[:7] if h else "HEAD"
	title = f"Shortlog - {project} @ {ref_display}"
	return _table_page(
		title=title,
		cols=["Commit", "Subject", "Author", "Age", "Diff"],
		rows=rows,
		pagination=_pagination_context(request, page, pagecount, page > 1, has_next),
	)


def git_history(
	project: str,
	h: str | None,
	f: str | None,
	request: Request,
	page: int,
	pagecount: int,
) -> HTMLResponse:
	"""History page for a file or directory. Port of git_history."""
	if not f:
		raise HTTPException(status_code=400, detail="File path (f) required for history")
	if not is_valid_pathname(f):
		raise HTTPException(status_code=400, detail="Invalid path")
	skip = (page - 1) * pagecount
	commits = get_commit_history(project, ref=h, path=f, max_count=pagecount + 1, skip=skip)
	if not commits:
		raise HTTPException(status_code=404, detail="No history found for this path")
	has_next = len(commits) > pagecount
	if has_next:
		commits = commits[:pagecount]
	rows = []
	for commit in commits:
		rows.append(_format_commit_table_row(project, commit, short=False))
	ref_display = h[:7] if h else "HEAD"
	title = f"History - {f} - {project} @ {ref_display}"
	return _table_page(
		title=title,
		project=project,
		cols=["Commit", "Subject", "Author", "Date", "Age", "Patch"],
		rows=rows,
		pagination=_pagination_context(request, page, pagecount, page > 1, has_next),
	)


def git_heads(project: str, token: str | None) -> HTMLResponse:
	"""Heads (branches) list page. Port of git_heads."""
	heads_list = git_get_heads_list(project)
	if settings.AUTH:
		visible_heads = [(name, ref, oid) for name, ref, oid in heads_list if can_read_branch(project, ref, token)]
	else:
		visible_heads = heads_list
	if not visible_heads:
		raise HTTPException(status_code=404, detail="No heads found")
	rows: list[list[Cell]] = []
	for name, _ref, oid in visible_heads:
		oid_short = oid[:7] if oid else ""
		commit_href = f"/project/{project}?a=commit&h={quote(oid, safe='')}"
		tree_href = f"/project/{project}?a=tree&h={quote(oid, safe='')}"
		rows.append([
			html_link(commit_href, name),
			html_link(commit_href, oid_short),
			html_link(tree_href, "tree"),
		])
	title = f"Heads - {project}"
	return _table_page(
		title=title,
		project=project,
		cols=["Head", "Commit", ""],
		rows=rows,
	)


def git_tags(project: str, request: Request, page: int, pagecount: int) -> HTMLResponse:
	"""Tags list page. Port of git_tags."""
	tags_list = git_get_tags_list(project)
	if not tags_list:
		raise HTTPException(status_code=404, detail="No tags found")
	total = len(tags_list)
	total_pages = (total + pagecount - 1) // pagecount if pagecount else 1
	if page > total_pages and total > 0:
		raise HTTPException(status_code=404, detail="Page does not exist")
	skip = (page - 1) * pagecount
	tags_page = tags_list[skip : skip + pagecount]
	try:
		repo = open_repo(project)
	except Exception as e:
		raise HTTPException(status_code=404, detail="Repository not found") from e
	rows = []
	for name, _ref, oid in tags_page:
		tag_link = f"/project/{project}?a=tag&h={quote(oid, safe='')}"
		target_oid = oid
		target_type = "commit"
		try:
			obj = repo.revparse_single(oid)
			if isinstance(obj, pygit2.Tag):
				target_oid = str(obj.target)
				target_type = obj.type_str
		except (KeyError, pygit2.GitError):
			pass
		target_short = target_oid[:7] if target_oid else ""
		if target_type == "commit":
			target_link = f"/project/{project}?a=commit&h={quote(target_oid, safe='')}"
		elif target_type == "tree":
			target_link = f"/project/{project}?a=tree&h={quote(target_oid, safe='')}"
		else:
			target_link = None
		obj_cell: Cell = html_link(target_link, target_short) if target_link else target_short
		rows.append([
			html_link(tag_link, name),
			obj_cell,
		])
	title = f"Tags - {project}"
	return _table_page(
		title=title,
		project=project,
		cols=["Tag", "Object"],
		rows=rows,
		pagination=_pagination_context(request, page, pagecount, page > 1, page < total_pages, total_pages=total_pages),
	)


def git_tag(project: str, h: str | None, notice: str | None = None) -> HTMLResponse | RedirectResponse:
	"""Single tag view. Port of git_tag."""
	if not h:
		raise HTTPException(status_code=400, detail="Tag ref or hash (h) required")
	if not is_valid_ref_format(h):
		raise HTTPException(status_code=400, detail="Invalid ref or hash")
	try:
		repo = open_repo(project)
		obj = repo.revparse_single(h)
	except (KeyError, pygit2.GitError, OSError) as e:
		raise HTTPException(status_code=404, detail="Tag or object not found") from e
	if not isinstance(obj, pygit2.Tag):
		raise HTTPException(status_code=404, detail="Not a tag object")
	mr_pair = resolve_merge_request_tag_to_tip(repo, obj)
	if mr_pair is not None:
		obj, mr = mr_pair
	else:
		mr = None
	tag_oid = str(obj.id)
	tag_data = parse_tag(project, tag_oid)
	if not tag_data:
		raise HTTPException(status_code=404, detail="Tag not found")
	target_oid = tag_data.get("object", "")
	target_type = tag_data.get("type", "commit")
	tagger = tag_data.get("tagger", "")
	tagger_epoch = tag_data.get("tagger_epoch")
	tagger_tz = tag_data.get("tagger_tz")
	message = (tag_data.get("message") or "").strip()
	tag_name = None
	for name, _ref, oid in git_get_tags_list(project):
		if oid == tag_oid:
			tag_name = name
			break
	if tag_name is None:
		tag_name = tag_oid[:7]
	if mr is not None:
		return merge_request_tag_response(
			project=project,
			repo=repo,
			mr=mr,
			tag_name=tag_name,
			tag_oid=tag_oid,
			tagger=tagger,
			tagger_date=_format_date(tagger_epoch, tagger_tz),
			notice=notice,
		)
	target_short = target_oid[:7] if target_oid else ""
	if target_type == "commit":
		object_value: Cell = html_link(
			f"/project/{project}?a=commit&h={quote(target_oid, safe='')}",
			target_short,
		)
	elif target_type == "tree":
		object_value = html_link(
			f"/project/{project}?a=tree&h={quote(target_oid, safe='')}",
			target_short,
		)
	else:
		object_value = target_short
	table_rows: list[list[Cell]] = [
		["Tag", tag_name or ""],
		["Object", html_join([object_value, " (", target_type, ")"])],
		["Tagger", tagger],
		["Date", _format_date(tagger_epoch, tagger_tz)],
	]
	if message:
		table_rows.append(["Message", html_format("<pre class='tag-message'>{}</pre>", message)])
	return _table_page(
		title=f"Tag {tag_name} - {project}",
		heading=f"Tag {tag_name}",
		project=project,
		cols=["Field", "Value"],
		rows=table_rows,
	)


def git_commit(project: str, h: str | None) -> HTMLResponse:
	"""Single commit view. Port of git_commit (git show)."""
	if not h:
		raise HTTPException(status_code=400, detail="Commit hash (h) required")
	if not is_valid_ref_format(h):
		raise HTTPException(status_code=400, detail="Invalid ref or hash")
	try:
		repo = open_repo(project)
		obj = repo.revparse_single(h)
	except (KeyError, pygit2.GitError, OSError) as e:
		raise HTTPException(status_code=404, detail="Commit not found") from e
	if not isinstance(obj, pygit2.Commit):
		raise HTTPException(status_code=404, detail="Not a commit object")
	commit = obj
	oid = str(commit.id)
	oid_short = oid[:7]
	data = parse_commit(project, oid)
	author = data.get("author", "")
	author_email = data.get("author_email", "")
	author_date = _format_date(data.get("author_epoch"), data.get("author_tz"))
	tree_oid = data.get("tree", "")
	message = (data.get("body") or "").strip()
	commit_link = f"/project/{project}?a=commit&h={quote(oid, safe='')}"
	tree_link = f"/project/{project}?a=tree&h={quote(oid, safe='')}"
	diff_link = f"/project/{project}?a=commitdiff&h={quote(oid, safe='')}"
	patch_link = f"/project/{project}?a=patch&h={quote(oid, safe='')}"
	table_rows: list[list[Cell]] = [
		["commit", html_link(commit_link, oid)],
		["Author", html_join([author, " <", author_email, ">"])],
		["Date", author_date],
		["tree", html_link(tree_link, tree_oid[:7])],
	]
	for i, parent_id in enumerate(commit.parent_ids):
		parent_oid = str(parent_id)
		parent_short = parent_oid[:7]
		parent_href = f"/project/{project}?a=commit&h={quote(parent_oid, safe='')}"
		label = "parent" if len(commit.parent_ids) == 1 else f"parent ({i + 1})"
		table_rows.append([label, html_link(parent_href, parent_short)])
	if message:
		table_rows.append(["", html_format("<pre class='commit-message'>{}</pre>", message)])
	table_rows.append(["", html_join([html_link(diff_link, "diff"), " · ", html_link(patch_link, "patch")])])
	return _table_page(
		title=f"Commit {oid_short} - {project}",
		heading=f"Commit {oid_short}",
		project=project,
		cols=["Field", "Value"],
		rows=table_rows,
	)


def git_commitdiff(project: str, h: str | None) -> HTMLResponse:
	"""Commit diff page: unified diff rendered with diff2html."""
	diff_text, oid_short = _commit_unified_diff_or_raise(project, h or "")
	diff_b64 = base64.b64encode(diff_text.encode("utf-8")).decode("ascii")
	return render_page(
		"commitdiff.html",
		title=f"Commit diff {oid_short} - {project}",
		project=project,
		oid_short=oid_short,
		diff_b64=diff_b64,
		commit_link=f"/project/{project}?a=commit&h={quote(h or '', safe='')}",
	)


def git_patch(project: str, h: str | None) -> PlainTextResponse:
	"""Single-commit patch (plain text unified diff)."""
	if not h:
		raise HTTPException(status_code=400, detail="Commit hash (h) required")
	body, oid_short = _commit_unified_diff_or_raise(project, h)
	filename = f"{project}-{oid_short}.patch"
	return PlainTextResponse(
		body,
		media_type="text/x-diff; charset=utf-8",
		headers={"Content-Disposition": f'inline; filename="{filename}"'},
	)


def git_patches(project: str, h: str | None, hb: str | None) -> PlainTextResponse:
	"""Multi-commit patches (plain text). Range hb..h if hb given, else single commit."""
	if not h:
		raise HTTPException(status_code=400, detail="Commit hash (h) required")
	if hb:
		oids = get_commits_in_range(project, h, hb)
		if not oids:
			raise HTTPException(status_code=404, detail="No commits in range")
		parts = []
		for oid in oids:
			try:
				diff, _ = get_commit_unified_diff(project, oid)
				if diff:
					parts.append(diff)
			except (KeyError, pygit2.GitError, OSError, ValueError):
				pass
		body = "\n".join(parts)
		filename = f"{project}-{h[:7]}-patches.patch"
	else:
		body, oid_short = _commit_unified_diff_or_raise(project, h)
		filename = f"{project}-{oid_short}.patch"
	return PlainTextResponse(
		body,
		media_type="text/x-diff; charset=utf-8",
		headers={"Content-Disposition": f'inline; filename="{filename}"'},
	)


def git_blobdiff(
	project: str,
	h: str | None,
	hb: str | None,
	f: str | None,
	fp: str | None,
) -> HTMLResponse:
	"""Blob diff page: diff between two blob versions, rendered with diff2html."""
	if not h or not hb:
		raise HTTPException(status_code=400, detail="Both h and hb (refs) required for blob diff")
	if not f:
		raise HTTPException(status_code=400, detail="File path (f) required for blob diff")
	if not is_valid_ref_format(h) or not is_valid_ref_format(hb):
		raise HTTPException(status_code=400, detail="Invalid ref or hash")
	path_new = f or ""
	path_old = fp if fp is not None else path_new
	if not is_valid_pathname(path_old) or not is_valid_pathname(path_new):
		raise HTTPException(status_code=400, detail="Invalid path")
	diff_text = get_blob_unified_diff(project, hb, path_old, h, path_new)
	if diff_text is None:
		raise HTTPException(status_code=404, detail="Blob not found")
	diff_b64 = base64.b64encode(diff_text.encode("utf-8")).decode("ascii")
	return render_page(
		"blobdiff.html",
		title=f"Blob diff - {path_new} - {project}",
		project=project,
		path_new=path_new,
		diff_b64=diff_b64,
		blob_link=f"/project/{project}?a=blob&h={quote(h or '', safe='')}&f={quote(path_new, safe='/')}",
	)


def git_blobpatch(
	project: str,
	h: str | None,
	hb: str | None,
	f: str | None,
	fp: str | None,
) -> PlainTextResponse:
	"""Blob diff as plain unified diff."""
	if not h or not hb:
		raise HTTPException(status_code=400, detail="Both h and hb (refs) required for blob diff")
	if not f:
		raise HTTPException(status_code=400, detail="File path (f) required for blob diff")
	if not is_valid_ref_format(h) or not is_valid_ref_format(hb):
		raise HTTPException(status_code=400, detail="Invalid ref or hash")
	path_new = f or ""
	path_old = fp if fp is not None else path_new
	if not is_valid_pathname(path_old) or not is_valid_pathname(path_new):
		raise HTTPException(status_code=400, detail="Invalid path")
	body = get_blob_unified_diff(project, hb, path_old, h, path_new)
	if body is None:
		raise HTTPException(status_code=404, detail="Blob not found")
	filename = f"{path_new.split('/')[-1] or 'blob'}.patch"
	return PlainTextResponse(
		body,
		media_type="text/x-diff; charset=utf-8",
		headers={"Content-Disposition": f'inline; filename="{filename}"'},
	)