@@ -0,0 +1,359 @@
#!/usr/bin/env python3
""" gitea-bot — mochi 全站 issue/PR 自动处理。
轮询 famzheng.me/gitea 所有 repo 的 open issue/PR；
每条 " 最新活动不是 mochi 自己 " 的条目，spawn 一个
`claude --dangerously-skip-permissions -p <prompt>`，
让 claude 用 mochi token 直接调 gitea API + git push 来回复 / 提 PR。
"""
import json
import os
import shlex
import sqlite3
import subprocess
import sys
import time
import urllib . error
import urllib . parse
import urllib . request
from datetime import datetime , timezone
from pathlib import Path
CONFIG = {
" base_url " : " https://famzheng.me/gitea " ,
" username " : " mochi " ,
" token_file " : Path . home ( ) / " .gitea-mochi-token " ,
" state_dir " : Path . home ( ) / " .local/state/gitea-bot " ,
" poll_interval_sec " : 60 ,
" claude_bin " : " /home/fam/.local/bin/claude " ,
" claude_timeout_sec " : 1200 ,
}
def log ( msg : str ) - > None :
ts = datetime . now ( ) . isoformat ( timespec = " seconds " )
print ( f " [ { ts } ] { msg } " , flush = True )
def load_token ( ) - > str :
return CONFIG [ " token_file " ] . read_text ( ) . strip ( )
def gitea_get ( path : str , token : str , * * params ) - > object :
url = f " { CONFIG [ ' base_url ' ] } /api/v1 { path } "
if params :
url + = " ? " + urllib . parse . urlencode ( params )
req = urllib . request . Request (
url ,
headers = {
" Authorization " : f " token { token } " ,
" Accept " : " application/json " ,
" User-Agent " : " gitea-bot/mochi " ,
} ,
)
with urllib . request . urlopen ( req , timeout = 30 ) as resp :
return json . loads ( resp . read ( ) )
def list_all_repos ( token : str ) - > list [ dict ] :
repos : list [ dict ] = [ ]
page = 1
while True :
batch = gitea_get ( " /repos/search " , token , limit = 50 , page = page )
items = batch . get ( " data " , [ ] ) if isinstance ( batch , dict ) else [ ]
if not items :
break
repos . extend ( items )
if len ( items ) < 50 :
break
page + = 1
return repos
def list_open_issues ( owner : str , repo : str , token : str , type_ : str ) - > list [ dict ] :
""" type_ in { ' issues ' , ' pulls ' } """
items : list [ dict ] = [ ]
page = 1
while True :
batch = gitea_get (
f " /repos/ { owner } / { repo } /issues " ,
token ,
state = " open " ,
type = type_ ,
page = page ,
limit = 50 ,
)
if not isinstance ( batch , list ) or not batch :
break
items . extend ( batch )
if len ( batch ) < 50 :
break
page + = 1
return items
def list_comments ( owner : str , repo : str , number : int , token : str ) - > list [ dict ] :
out = gitea_get ( f " /repos/ { owner } / { repo } /issues/ { number } /comments " , token )
return out if isinstance ( out , list ) else [ ]
def init_state ( db_path : Path ) - > sqlite3 . Connection :
db_path . parent . mkdir ( parents = True , exist_ok = True )
conn = sqlite3 . connect ( db_path )
conn . execute (
"""
CREATE TABLE IF NOT EXISTS handled (
key TEXT PRIMARY KEY,
handled_at TEXT NOT NULL,
last_signal TEXT NOT NULL
)
"""
)
conn . commit ( )
return conn
def get_handled_signal ( conn : sqlite3 . Connection , key : str ) - > str | None :
row = conn . execute (
" SELECT last_signal FROM handled WHERE key = ? " , ( key , )
) . fetchone ( )
return row [ 0 ] if row else None
def set_handled ( conn : sqlite3 . Connection , key : str , signal : str ) - > None :
conn . execute (
"""
INSERT INTO handled(key, handled_at, last_signal) VALUES (?, ?, ?)
ON CONFLICT(key) DO UPDATE SET
handled_at = excluded.handled_at,
last_signal = excluded.last_signal
""" ,
( key , datetime . now ( timezone . utc ) . isoformat ( ) , signal ) ,
)
conn . commit ( )
def latest_signal ( issue : dict , comments : list [ dict ] ) - > tuple [ str , str ] :
""" Return (time, author_login) for the latest activity on this issue. """
cands : list [ tuple [ str , str ] ] = [
( issue [ " updated_at " ] , issue [ " user " ] [ " login " ] ) ,
]
for c in comments :
cands . append ( ( c [ " updated_at " ] , c [ " user " ] [ " login " ] ) )
cands . sort ( key = lambda x : x [ 0 ] )
return cands [ - 1 ]
def build_prompt ( repo : dict , issue : dict , is_pr : bool , comments : list [ dict ] ) - > str :
typ = " pull request " if is_pr else " issue "
full = repo [ " full_name " ]
labels = " , " . join ( l [ " name " ] for l in issue . get ( " labels " ) or [ ] ) or " (none) "
body = ( issue . get ( " body " ) or " (empty) " ) . strip ( )
if len ( body ) > 8000 :
body = body [ : 8000 ] + " \n …(truncated) "
comments_tail = comments [ - 10 : ]
if comments_tail :
comments_str = " \n \n " . join (
f " ### @ { c [ ' user ' ] [ ' login ' ] } at { c [ ' created_at ' ] } \n { ( c . get ( ' body ' ) or ' ' ) . strip ( ) } "
for c in comments_tail
)
else :
comments_str = " (no comments yet) "
pr_note = " "
if is_pr :
pr_note = (
f " \n This is a PULL REQUEST. You can fetch the diff at "
f " `GET /repos/ { full } /pulls/ { issue [ ' number ' ] } .diff` (raw text). "
f " You may post review comments via the issues comments endpoint, "
f " or full reviews via `POST /repos/ { full } /pulls/ { issue [ ' number ' ] } /reviews`. \n "
)
return f """ You are **mochi** (麻薯), the autonomous Gitea bot running site-wide on
{ CONFIG [ ' base_url ' ] } . You are invoked once per " new activity " on an issue or PR
and must finish your work in this single run.
# Identity & access
- Gitea username: `mochi` (you have admin on this instance).
- Auth token: env var `MOCHI_TOKEN`. Send it as `Authorization: token $MOCHI_TOKEN`.
- API base: ` { CONFIG [ ' base_url ' ] } /api/v1`
- SSH is disabled — git over HTTPS only. To clone/push:
`git clone https://mochi:$MOCHI_TOKEN@famzheng.me/gitea/<owner>/<repo>.git`
- Configure git author when committing:
`git -c user.name=mochi -c user.email=mochi@famzheng.me commit ...`
- Work in `/tmp/gitea-bot-<rand>/`. Clean up when done.
- You have full shell access (--dangerously-skip-permissions). Be careful but decisive.
# Voice
- Smart, concise senior-engineer tone. No customer-service fluff.
- Reply in the **same language** as the issue body (中文 or English — detect it).
- Do NOT use bracketed RP actions like `(笑)` `(歪头)` `(撒娇)`. Express tone with words.
- Sign every comment you POST with this footer line on its own line:
`🤖 mochi bot · automated · { CONFIG [ ' base_url ' ] } `
# Target { typ }
- Repo: ` { full } ` (default branch: ` { repo . get ( ' default_branch ' , ' main ' ) } `)
- Issue/PR #: { issue [ ' number ' ] }
- Title: { issue [ ' title ' ] }
- Author: @ { issue [ ' user ' ] [ ' login ' ] }
- URL: { issue [ ' html_url ' ] }
- Labels: { labels }
- State: { issue [ ' state ' ] }
- Created: { issue [ ' created_at ' ] } · Updated: { issue [ ' updated_at ' ] }
## Body
{ body }
## Recent comments (oldest → newest, up to 10)
{ comments_str }
{ pr_note }
# What to do
1. Re-read the issue/PR. If the **latest** comment is from `mochi` and there ' s no
new question/instruction after it, exit silently — do not double-post.
2. Classify and act:
- **Question / discussion / unclear ask** → post **one** helpful comment
(≤ 8 lines). Ask clarifying questions if needed. Do not open a PR.
- **Feature request** (clear, scoped, small enough that you can implement it
in one PR):
a. `git clone` the repo to a temp dir.
b. `git switch -c mochi/issue- { issue [ ' number ' ] } `
c. Implement the change. If the repo has obvious build/test commands
(Makefile, package.json scripts, cargo, etc.), run them and ensure they pass.
d. Commit with a clear message; author `mochi <mochi@famzheng.me>`.
e. Push the branch.
f. Open a PR via `POST /repos/ { full } /pulls` with base = default branch,
head = `mochi/issue- { issue [ ' number ' ] } `, title like
`feat: <summary> (closes # { issue [ ' number ' ] } )`, body referencing the issue.
g. Post **one** short comment on the issue linking the PR.
If the request is too large, ambiguous, or risky → instead post a comment
with a proposed plan and ask for confirmation. Don ' t push.
- **Bug report** → if the fix is obvious from reading the code, follow the
same flow as feature request with title `fix: <summary> (closes # { issue [ ' number ' ] } )`.
Otherwise post a diagnostic comment asking for repro steps / logs.
- **Pull request** → if the PR author is `mochi`, exit. Otherwise fetch the
diff and post **one** brief review comment (correctness/style/obvious bugs).
Don ' t push commits to other people ' s PRs.
3. Be conservative. When unsure, comment instead of pushing code.
Begin now. When done, exit.
"""
def dispatch (
repo : dict , issue : dict , is_pr : bool , comments : list [ dict ] , token : str
) - > None :
prompt = build_prompt ( repo , issue , is_pr , comments )
env = os . environ . copy ( )
env [ " MOCHI_TOKEN " ] = token
# ensure claude can find git, curl, node (claude wraps node), etc.
env [ " PATH " ] = (
" /home/fam/.local/bin:/home/fam/.cargo/bin: "
" /usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin "
)
env [ " HOME " ] = " /home/fam "
full = repo [ " full_name " ]
log (
f " dispatch { full } # { issue [ ' number ' ] } "
f " ( { ' PR ' if is_pr else ' issue ' } ): { issue [ ' title ' ] !r} "
)
cmd = [ CONFIG [ " claude_bin " ] , " --dangerously-skip-permissions " , " -p " , prompt ]
try :
r = subprocess . run (
cmd ,
env = env ,
capture_output = True ,
text = True ,
timeout = CONFIG [ " claude_timeout_sec " ] ,
cwd = " /tmp " ,
)
log ( f " → exit= { r . returncode } , stdout= { len ( r . stdout ) } B, stderr= { len ( r . stderr ) } B " )
if r . stdout . strip ( ) :
tail = r . stdout [ - 1500 : ]
log ( f " stdout tail: \n { tail } " )
if r . stderr . strip ( ) :
tail = r . stderr [ - 1500 : ]
log ( f " stderr tail: \n { tail } " )
except subprocess . TimeoutExpired :
log ( f " → TIMEOUT after { CONFIG [ ' claude_timeout_sec ' ] } s " )
except Exception as e :
log ( f " → dispatch error: { e !r} " )
def poll_once ( conn : sqlite3 . Connection , token : str ) - > None :
repos = list_all_repos ( token )
log ( f " polling { len ( repos ) } repos " )
for repo in repos :
if repo . get ( " archived " ) :
continue
owner = repo [ " owner " ] [ " username " ] if " username " in repo [ " owner " ] else repo [ " owner " ] [ " login " ]
name = repo [ " name " ]
full = repo [ " full_name " ]
for type_ in ( " issues " , " pulls " ) :
try :
items = list_open_issues ( owner , name , token , type_ )
except urllib . error . HTTPError as e :
if e . code in ( 403 , 404 ) :
continue
log ( f " ! list { full } { type_ } : HTTP { e . code } " )
continue
except Exception as e :
log ( f " ! list { full } { type_ } : { e !r} " )
continue
for issue in items :
is_pr = " pull_request " in issue and issue [ " pull_request " ] is not None
if type_ == " issues " and is_pr :
continue
if type_ == " pulls " and not is_pr :
continue
num = issue [ " number " ]
try :
comments = list_comments ( owner , name , num , token )
except Exception as e :
log ( f " ! comments { full } # { num } : { e !r} " )
comments = [ ]
latest_time , latest_user = latest_signal ( issue , comments )
key = f " { full } # { num } "
if latest_user == CONFIG [ " username " ] :
# last activity is mochi's own — record & skip
set_handled ( conn , key , latest_time )
continue
if get_handled_signal ( conn , key ) == latest_time :
continue
dispatch ( repo , issue , is_pr , comments , token )
set_handled ( conn , key , latest_time )
def main ( ) - > None :
state_dir = CONFIG [ " state_dir " ]
state_dir . mkdir ( parents = True , exist_ok = True )
conn = init_state ( state_dir / " state.db " )
log (
f " gitea-bot up; base= { CONFIG [ ' base_url ' ] } "
f " interval= { CONFIG [ ' poll_interval_sec ' ] } s "
f " claude= { CONFIG [ ' claude_bin ' ] } "
)
while True :
try :
token = load_token ( )
poll_once ( conn , token )
except Exception as e :
log ( f " poll error: { e !r} " )
time . sleep ( CONFIG [ " poll_interval_sec " ] )
if __name__ == " __main__ " :
main ( )
Fam Zheng