okooooooo/biliup
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(stream-gears): Python signatures, types and error semantics (#1729)

Browse Source 
* fix(stream-gears): raise exceptions instead of failing silently in the Python API

- Add `stream_gears.StreamGearsError` (subclass of `RuntimeError`, so
  existing `except RuntimeError` keeps working). Errors that used to raise
  `RuntimeError` now raise it, with the full error-stack cause chain.
- `download` / `download_with_callback`: connection failures, an
  incomplete first packet, FLV errors mid-stream and HLS errors raise
  `StreamGearsError` instead of panicking (`PanicException` derives from
  `BaseException`) or only logging and returning. The FLV path calls
  `httpflv::parse_flv`, now public, so `httpflv::download` and server-side
  recording are unchanged.
- `download_with_callback` accepts any callable. A callback exception is
  logged when it happens and re-raised once the download finishes; if the
  download fails too, it becomes the `__cause__` of the `StreamGearsError`.
- `PySegment` fields are readable; `segment` also accepts a dict or an
  object with `time` / `size` attributes, and `desc_v2` items accept
  objects with `type_id` / `raw_text` / `biz_id` besides dicts.
- `upload`: malformed `extra_fields` raises `ValueError` instead of being
  dropped; empty or blank still means none.
- `login_by_sms` raises on failure instead of returning `False`; invalid
  `ret` JSON raises `ValueError`; tokio runtime creation errors raise
  instead of panicking.
- The seven login functions get `proxy=None` defaults (PyO3 >= 0.24 no
  longer defaults trailing `Option` arguments).

Ref #1721

* test(stream-gears): offline pytest cases for error semantics and argument types

Ref #1721

* fix(stream-gears): turn FLV parser panics in download into StreamGearsError

`httpflv::parse_flv` still `expect`s on malformed AAC/AVC/script tag data.
Catch the unwind at the binding boundary so `download` raises
`StreamGearsError` instead of `PanicException`, without changing the
parser that server-side recording shares.

Ref #1721

* test(stream-gears): update smoke suite for StreamGearsError and optional proxy

The login functions now default proxy to None, so the strict xfail cases
for it pass and are removed (test_errors.py covers the same assertion).
StreamGearsError joins the exported names; PySegment fields are readable.

* feat(stream-gears): optional cookie file path for login functions

`login_by_sms`, `login_by_web_cookies` and `login_by_web_qrcode` take a
`file` argument (default `"cookies.json"`, same as before) for where the
login info is written. `login_by_qrcode` takes `file=None` and writes the
login info there when given; it still returns the JSON string.

Ref #1721
This commit is contained in:
Noah authored and GitHub committed 2026-09-24 05:27:39 +00:00
1 parent 753aade9f4
commit 55a260f9da
6 files changed
+623 -167
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/biliup/src/downloader/httpflv.rs
+2 -1
View File
@@ -33,7 +33,8 @@ pub async fn download(
}
}
pub(crate) async fn parse_flv(
/// 同 [`download`]，但把错误返回给调用方，而不是只打一条 `warn` 日志。
pub async fn parse_flv(
mut connection: Connection,
file: LifecycleFile<'_>,
mut segment: Segmentable,
crates/stream-gears/src/lib.rs
+290 -136
View File
@@ -11,9 +11,13 @@ use axum::http::HeaderMap;
use biliup::credential::Credential;
use biliup::downloader::util::{CallbackFn, LifecycleFile, Segmentable};
use biliup::downloader::{hls, httpflv};
use pyo3::types::{PyList, PyType};
use biliup::uploader::credential::save_login_info;
use pyo3::types::{PyList, PyMapping};
use std::collections::HashMap;
use std::fmt::Display;
use std::panic::{self, AssertUnwindSafe};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tracing::{debug, error, info};
@@ -21,56 +25,103 @@ use biliup::client::StatelessClient;
use biliup::downloader::flv_parser::header;
use biliup::downloader::httpflv::Connection;
use biliup_cli::server::common::construct_headers;
use pyo3::exceptions::{PyRuntimeError, PySystemExit};
use pyo3::exceptions::{PyRuntimeError, PySystemExit, PyTypeError, PyValueError};
use tracing_subscriber::layer::SubscriberExt;
#[tokio::main]
pyo3::create_exception!(
stream_gears,
StreamGearsError,
PyRuntimeError,
"Raised when a download, login or upload fails. Subclass of RuntimeError."
);
/// `{:#}` 让 `error_stack::Report` 带上整条原因链，而不只是最外层的 "Unknown Error"。
fn stream_gears_err(err: impl Display) -> PyErr {
StreamGearsError::new_err(format!("{err:#}"))
}
fn new_runtime() -> PyResult<tokio::runtime::Runtime> {
tokio::runtime::Runtime::new()
.map_err(|e| StreamGearsError::new_err(format!("failed to start tokio runtime: {e}")))
}
fn parse_json_arg(name: &str, value: &str) -> PyResult<serde_json::Value> {
serde_json::from_str(value)
.map_err(|e| PyValueError::new_err(format!("{name} is not valid JSON: {e}")))
}
/// 从 mapping 里按键、从其它对象上按属性取字段；字段不存在时返回 `None`。
pub(crate) fn get_field<'py>(
obj: &Bound<'py, PyAny>,
key: &str,
) -> PyResult<Option<Bound<'py, PyAny>>> {
if let Ok(mapping) = obj.cast::<PyMapping>() {
return if mapping.contains(key)? {
mapping.get_item(key).map(Some)
} else {
Ok(None)
};
}
if obj.hasattr(key)? {
obj.getattr(key).map(Some)
} else {
Ok(None)
}
}
pub async fn download_with_hook(
url: &str,
headers: HeaderMap,
file_name: &str,
segment: Segmentable,
file_name_hook: CallbackFn,
file_name_hook: CallbackFn<'_>,
proxy: Option<&str>,
) -> PyResult<()> {
let client = StatelessClient::new(headers, proxy);
let response = client.retryable(url).await.unwrap();
let response = client
.retryable(url)
.await
.map_err(|e| StreamGearsError::new_err(format!("failed to connect to {url}: {e}")))?;
let mut connection = Connection::new(response);
// let buf = &mut [0u8; 9];
let bytes = connection.read_frame(9).await.unwrap();
// response.read_exact(buf)?;
// let out = File::create(format!("{}.flv", file_name)).expect("Unable to create file.");
// let mut writer = BufWriter::new(out);
// let mut buf = [0u8; 8 * 1024];
// response.copy_to(&mut writer)?;
// io::copy(&mut resp, &mut out).expect("Unable to copy the content.");
let bytes = connection
.read_frame(9)
.await
.map_err(|e| StreamGearsError::new_err(format!("failed to read from {url}: {e}")))?;
match header(&bytes) {
Ok((_i, header)) => {
debug!("header: {header:#?}");
info!("Downloading {}...", url);
let file = LifecycleFile::with_hook(file_name, "flv", file_name_hook);
httpflv::download(connection, file, segment, None).await;
httpflv::parse_flv(connection, file, segment, None)
.await
.map_err(|e| {
StreamGearsError::new_err(format!("FLV download from {url} failed: {e}"))
})?;
info!("Done... {}", file_name);
}
Err(nom::Err::Incomplete(needed)) => {
error!("needed: {needed:?}")
return Err(StreamGearsError::new_err(format!(
"incomplete FLV header from {url}: got {} bytes, needed {needed:?} more",
bytes.len()
)));
}
Err(e) => {
error!("{e}");
let file = LifecycleFile::with_hook(file_name, "ts", file_name_hook);
hls::download(url, &client, file, segment, None)
.await
.unwrap();
.map_err(|e| {
StreamGearsError::new_err(format!("HLS download from {url} failed: {e}"))
})?;
}
}
Ok(())
}
#[derive(Debug, Clone)]
#[pyclass(set_all)]
#[pyclass(get_all, set_all)]
pub struct PySegment {
// #[pyo3(attribute("time"))]
time: Option<u64>,
// #[pyo3(attribute("size"))]
size: Option<u64>,
}
@@ -85,6 +136,44 @@ impl PySegment {
}
}
/// `download` 的 `segment` 参数：`PySegment`、带 `time` / `size` 键的 dict，
/// 或带 `time` / `size` 属性的对象。
struct SegmentArg {
time: Option<u64>,
size: Option<u64>,
}
impl<'a, 'py> FromPyObject<'a, 'py> for SegmentArg {
type Error = PyErr;
fn extract(obj: Borrowed<'a, 'py, PyAny>) -> PyResult<Self> {
if let Ok(segment) = obj.cast::<PySegment>() {
let segment = segment.borrow();
return Ok(SegmentArg {
time: segment.time,
size: segment.size,
});
}
let obj = obj.to_owned();
if obj.cast::<PyMapping>().is_err() && !(obj.hasattr("time")? || obj.hasattr("size")?) {
return Err(PyTypeError::new_err(format!(
"segment must be a PySegment, a dict or an object with `time` / `size` attributes, not {}",
obj.get_type().name()?
)));
}
let field = |key| -> PyResult<Option<u64>> {
Ok(get_field(&obj, key)?
.map(|value| value.extract::<Option<u64>>())
.transpose()?
.flatten())
};
Ok(SegmentArg {
time: field("time")?,
size: field("size")?,
})
}
}
#[pyfunction]
#[pyo3(signature = (url,header_map,file_name,segment,proxy = None))]
fn download(
@@ -92,12 +181,15 @@ fn download(
url: &str,
header_map: HashMap<String, String>,
file_name: &str,
segment: PySegment,
segment: SegmentArg,
proxy: Option<String>,
) -> PyResult<()> {
download_with_callback(py, url, header_map, file_name, segment, None, proxy)
}
/// 回调在每个分段文件写完时被调用。回调抛出的异常会立即记进日志，下载继续；
/// 下载结束后重新抛出第一个回调异常。下载本身也失败时抛 `StreamGearsError`，
/// 回调异常挂在它的 `__cause__` 上。
#[pyfunction]
#[pyo3(signature = (url,header_map,file_name,segment,file_name_callback_fn = None,proxy = None))]
fn download_with_callback(
@@ -105,12 +197,35 @@ fn download_with_callback(
url: &str,
header_map: HashMap<String, String>,
file_name: &str,
segment: PySegment,
file_name_callback_fn: Option<Py<PyType>>,
segment: SegmentArg,
file_name_callback_fn: Option<Py<PyAny>>,
proxy: Option<String>,
) -> PyResult<()> {
py.detach(|| {
let map = construct_headers(&header_map).map_err(PyRuntimeError::new_err)?;
if let Some(callback_fn) = &file_name_callback_fn
&& !callback_fn.bind(py).is_callable()
{
return Err(PyTypeError::new_err(format!(
"file_name_callback_fn must be callable, not {}",
callback_fn.bind(py).get_type().name()?
)));
}
let callback_error: Arc<Mutex<Option<PyErr>>> = Arc::default();
let file_name_hook = file_name_callback_fn.map(|callback_fn| -> CallbackFn {
let callback_error = Arc::clone(&callback_error);
Box::new(move |fmt_file_name| {
Python::attach(|py| {
if let Err(err) = callback_fn.call1(py, (fmt_file_name,)) {
error!("file_name_callback_fn({fmt_file_name:?}) raised {err}");
if let Ok(mut first) = callback_error.lock() {
first.get_or_insert(err);
}
}
})
})
});
let result = py.detach(|| {
let map = construct_headers(&header_map).map_err(StreamGearsError::new_err)?;
// 输出到控制台中
// use of deprecated function `time::util::local_offset::set_soundness`: no longer needed; TZ is refreshed manually
// unsafe {
@@ -131,9 +246,6 @@ fn download_with_callback(
.with_timer(local_time)
.with_writer(non_blocking);
// println!("Input segment: {:?}", segment);
// println!("Input segment time: {:?}, size: {:?}", segment.time, segment.size);
let segmentable = match (segment.time, segment.size) {
(Some(time), Some(size)) => {
// 已支持同时创建时间和大小
@@ -147,53 +259,63 @@ fn download_with_callback(
}
};
let file_name_hook = file_name_callback_fn.map(|callback_fn| -> CallbackFn {
Box::new(move |fmt_file_name| {
Python::attach(|py| match callback_fn.call1(py, (fmt_file_name,)) {
Ok(_) => {}
Err(_) => {
tracing::error!("Unable to invoke the callback function.")
}
})
})
});
let collector = formatting_layer.with(file_layer);
tracing::subscriber::with_default(collector, || -> PyResult<()> {
match download_with_hook(
url,
map,
file_name,
segmentable,
file_name_hook.unwrap_or(Box::new(|_| {})),
proxy.as_deref(),
) {
Ok(res) => Ok(res),
Err(err) => Err(pyo3::exceptions::PyRuntimeError::new_err(format!(
"{}",
err
))),
}
let rt = new_runtime()?;
// `httpflv::parse_flv` 遇到畸形的 tag 数据仍会 `expect` panic；不在这里拦下，
// Python 侧收到的是 `except Exception` 接不住的 PanicException。
panic::catch_unwind(AssertUnwindSafe(|| {
rt.block_on(download_with_hook(
url,
map,
file_name,
segmentable,
file_name_hook.unwrap_or(Box::new(|_| {})),
proxy.as_deref(),
))
}))
.unwrap_or_else(|payload| {
let reason = payload
.downcast_ref::<&str>()
.copied()
.or_else(|| payload.downcast_ref::<String>().map(String::as_str))
.unwrap_or("unknown panic");
Err(StreamGearsError::new_err(format!(
"download from {url} failed: {reason}"
)))
})
})
})
});
let callback_error = callback_error
.lock()
.ok()
.and_then(|mut first| first.take());
match (result, callback_error) {
(Ok(()), None) => Ok(()),
(Ok(()), Some(callback_error)) => Err(callback_error),
(Err(err), callback_error) => {
if callback_error.is_some() {
err.set_cause(py, callback_error);
}
Err(err)
}
}
}
#[pyfunction]
#[pyo3(signature = (file, proxy=None))]
fn login_by_cookies(py: Python<'_>, file: String, proxy: Option<String>) -> PyResult<bool> {
py.detach(|| {
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(async { login::login_by_cookies(&file, proxy.as_deref()).await });
match result {
Ok(_) => Ok(true),
Err(err) => Err(pyo3::exceptions::PyRuntimeError::new_err(format!(
"{}",
err
))),
}
new_runtime()?
.block_on(login::login_by_cookies(&file, proxy.as_deref()))
.map(|_| true)
.map_err(stream_gears_err)
})
}
#[pyfunction]
#[pyo3(signature = (country_code, phone, proxy=None))]
fn send_sms(
py: Python<'_>,
country_code: u32,
@@ -201,104 +323,108 @@ fn send_sms(
proxy: Option<String>,
) -> PyResult<String> {
py.detach(|| {
let rt = tokio::runtime::Runtime::new().unwrap();
let result =
rt.block_on(async { login::send_sms(country_code, phone, proxy.as_deref()).await });
match result {
Ok(res) => Ok(res.to_string()),
Err(err) => Err(pyo3::exceptions::PyRuntimeError::new_err(format!(
"{}",
err
))),
}
new_runtime()?
.block_on(login::send_sms(country_code, phone, proxy.as_deref()))
.map(|res| res.to_string())
.map_err(stream_gears_err)
})
}
/// 成功时返回 `True`，并把登录信息写入 `file`（默认当前目录的 `cookies.json`）。
/// 登录失败抛 `StreamGearsError`（消息里带原因），`ret` 不是合法 JSON 时抛 `ValueError`。
#[pyfunction]
fn login_by_sms(py: Python<'_>, code: u32, ret: String, proxy: Option<String>) -> PyResult<bool> {
#[pyo3(signature = (code, ret, proxy=None, file=PathBuf::from("cookies.json")))]
#[pyo3(text_signature = "(code, ret, proxy=None, file='cookies.json')")]
fn login_by_sms(
py: Python<'_>,
code: u32,
ret: String,
proxy: Option<String>,
file: PathBuf,
) -> PyResult<bool> {
let ret = parse_json_arg("ret", &ret)?;
py.detach(|| {
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(async {
login::login_by_sms(code, serde_json::from_str(&ret).unwrap(), proxy.as_deref()).await
});
match result {
Ok(_) => Ok(true),
Err(_) => Ok(false),
}
new_runtime()?
.block_on(login::login_by_sms(code, ret, proxy.as_deref(), &file))
.map_err(stream_gears_err)
})
}
#[pyfunction]
#[pyo3(signature = (proxy=None))]
fn get_qrcode(py: Python<'_>, proxy: Option<String>) -> PyResult<String> {
py.detach(|| {
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(async { login::get_qrcode(proxy.as_deref()).await });
match result {
Ok(res) => Ok(res.to_string()),
Err(err) => Err(pyo3::exceptions::PyRuntimeError::new_err(format!(
"{}",
err
))),
}
new_runtime()?
.block_on(login::get_qrcode(proxy.as_deref()))
.map(|res| res.to_string())
.map_err(stream_gears_err)
})
}
/// 返回登录信息的 JSON 字符串；给了 `file` 时同时写入该文件。
#[pyfunction]
fn login_by_qrcode(py: Python<'_>, ret: String, proxy: Option<String>) -> PyResult<String> {
#[pyo3(signature = (ret, proxy=None, file=None))]
fn login_by_qrcode(
py: Python<'_>,
ret: String,
proxy: Option<String>,
file: Option<PathBuf>,
) -> PyResult<String> {
let ret = parse_json_arg("ret", &ret)?;
py.detach(|| {
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let info = Credential::new(proxy.as_deref())
.login_by_qrcode(serde_json::from_str(&ret).unwrap())
.await?;
let res = serde_json::to_string_pretty(&info).unwrap();
Ok::<_, biliup::error::Kind>(res)
})
.map_err(|err| pyo3::exceptions::PyRuntimeError::new_err(format!("{:#?}", err)))
let rt = new_runtime()?;
let info = rt
.block_on(Credential::new(proxy.as_deref()).login_by_qrcode(ret))
.map_err(stream_gears_err)?;
if let Some(file) = &file {
rt.block_on(save_login_info(file, &info))
.map_err(stream_gears_err)?;
}
serde_json::to_string_pretty(&info).map_err(stream_gears_err)
})
}
#[pyfunction]
#[pyo3(signature = (sess_data, bili_jct, proxy=None, file=PathBuf::from("cookies.json")))]
#[pyo3(text_signature = "(sess_data, bili_jct, proxy=None, file='cookies.json')")]
fn login_by_web_cookies(
py: Python<'_>,
sess_data: String,
bili_jct: String,
proxy: Option<String>,
file: PathBuf,
) -> PyResult<bool> {
py.detach(|| {
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(async {
login::login_by_web_cookies(&sess_data, &bili_jct, proxy.as_deref()).await
});
match result {
Ok(_) => Ok(true),
Err(err) => Err(pyo3::exceptions::PyRuntimeError::new_err(format!(
"{}",
err
))),
}
new_runtime()?
.block_on(login::login_by_web_cookies(
&sess_data,
&bili_jct,
proxy.as_deref(),
&file,
))
.map_err(stream_gears_err)
})
}
#[pyfunction]
#[pyo3(signature = (sess_data, dede_user_id, proxy=None, file=PathBuf::from("cookies.json")))]
#[pyo3(text_signature = "(sess_data, dede_user_id, proxy=None, file='cookies.json')")]
fn login_by_web_qrcode(
py: Python<'_>,
sess_data: String,
dede_user_id: String,
proxy: Option<String>,
file: PathBuf,
) -> PyResult<bool> {
py.detach(|| {
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(async {
login::login_by_web_qrcode(&sess_data, &dede_user_id, proxy.as_deref()).await
});
match result {
Ok(_) => Ok(true),
Err(err) => Err(pyo3::exceptions::PyRuntimeError::new_err(format!(
"{}",
err
))),
}
new_runtime()?
.block_on(login::login_by_web_qrcode(
&sess_data,
&dede_user_id,
proxy.as_deref(),
&file,
))
.map_err(stream_gears_err)
})
}
@@ -333,6 +459,9 @@ fn upload(
submit: Option<String>,
proxy: Option<String>,
) -> PyResult<()> {
let extra_fields = parse_extra_fields(extra_fields.as_deref()).map_err(|e| {
PyValueError::new_err(format!("extra_fields is not a valid JSON object: {e}"))
})?;
py.detach(|| {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
@@ -383,7 +512,7 @@ fn upload(
.up_selection_reply(up_selection_reply)
.up_close_danmu(up_close_danmu)
.desc_v2_credit(desc_v2)
.extra_fields(parse_extra_fields(extra_fields))
.extra_fields(extra_fields)
.build();
// let submit = match submit {
@@ -391,18 +520,13 @@ fn upload(
// None => SubmitOption::App,
// };
match rt.block_on(uploader::upload(
rt.block_on(uploader::upload(
studio_pre,
submit.as_deref(),
proxy.as_deref(),
)) {
Ok(_) => Ok(()),
// Ok(_) => { },
Err(err) => Err(pyo3::exceptions::PyRuntimeError::new_err(format!(
"{}",
err
))),
}
))
.map(|_| ())
.map_err(stream_gears_err)
})
})
}
@@ -461,12 +585,42 @@ fn stream_gears(m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_function(wrap_pyfunction!(server::config_bindings, m)?)?;
m.add_class::<UploadLine>()?;
m.add_class::<PySegment>()?;
m.add("StreamGearsError", m.py().get_type::<StreamGearsError>())?;
Ok(())
}
fn parse_extra_fields(s: Option<String>) -> HashMap<String, serde_json::Value> {
match s {
Some(value) => serde_json::from_str(&value).unwrap_or_default(), // 如果有值，尝试解析
None => HashMap::new(), // 如果是 None，直接返回空的 HashMap
/// `None`、空串或只含空白视为没有额外字段；其余必须是 JSON 对象。
fn parse_extra_fields(s: Option<&str>) -> serde_json::Result<HashMap<String, serde_json::Value>> {
match s.map(str::trim) {
None | Some("") => Ok(HashMap::new()),
Some(value) => serde_json::from_str(value),
}
}
#[cfg(test)]
mod tests {
use super::parse_extra_fields;
use serde_json::json;
#[test]
fn missing_or_blank_extra_fields_are_empty() {
assert!(parse_extra_fields(None).unwrap().is_empty());
assert!(parse_extra_fields(Some("")).unwrap().is_empty());
assert!(parse_extra_fields(Some(" \n")).unwrap().is_empty());
}
#[test]
fn extra_fields_json_object_is_parsed() {
let fields = parse_extra_fields(Some(r#"{"a": 1, "b": {"c": "d"}}"#)).unwrap();
assert_eq!(fields.len(), 2);
assert_eq!(fields["a"], json!(1));
assert_eq!(fields["b"], json!({"c": "d"}));
}
#[test]
fn malformed_extra_fields_are_rejected() {
for input in [r#"{"a":1,}"#, "{", "[1, 2]", "null", "1", r#""a""#] {
assert!(parse_extra_fields(Some(input)).is_err(), "{input}");
}
}
}
crates/stream-gears/src/login.rs
+7 -3
View File
@@ -2,6 +2,7 @@ use biliup::uploader::bilibili::BiliBili;
use biliup::uploader::credential::{Credential, save_login_info};
use biliup_cli::server::errors::{AppError, AppResult};
use error_stack::ResultExt;
use std::path::Path;
pub async fn login_by_cookies(file: &str, proxy: Option<&str>) -> AppResult<BiliBili> {
let login_info = biliup::uploader::credential::login_by_cookies(file, proxy)
@@ -26,12 +27,13 @@ pub async fn login_by_sms(
code: u32,
res: serde_json::Value,
proxy: Option<&str>,
file: &Path,
) -> AppResult<bool> {
let info = Credential::new(proxy)
.login_by_sms(code, res)
.await
.change_context_lazy(|| AppError::Unknown)?;
save_login_info("cookies.json", &info)
save_login_info(file, &info)
.await
.change_context_lazy(|| AppError::Unknown)?;
Ok(true)
@@ -49,12 +51,13 @@ pub async fn login_by_web_cookies(
sess_data: &str,
bili_jct: &str,
proxy: Option<&str>,
file: &Path,
) -> AppResult<bool> {
let info = Credential::new(proxy)
.login_by_web_cookies(sess_data, bili_jct)
.await
.change_context_lazy(|| AppError::Unknown)?;
save_login_info("cookies.json", &info)
save_login_info(file, &info)
.await
.change_context_lazy(|| AppError::Unknown)?;
Ok(true)
@@ -64,12 +67,13 @@ pub async fn login_by_web_qrcode(
sess_data: &str,
dede_user_id: &str,
proxy: Option<&str>,
file: &Path,
) -> AppResult<bool> {
let info = Credential::new(proxy)
.login_by_web_qrcode(sess_data, dede_user_id)
.await
.change_context_lazy(|| AppError::Unknown)?;
save_login_info("cookies.json", &info)
save_login_info(file, &info)
.await
.change_context_lazy(|| AppError::Unknown)?;
Ok(true)
crates/stream-gears/src/uploader.rs
+30 -4
View File
@@ -1,7 +1,10 @@
use biliup::uploader::bilibili::{Credit, ResponseData, Studio};
use pyo3::exceptions::PyTypeError;
use pyo3::prelude::*;
use pyo3::pyclass;
use pyo3::types::PyMapping;
use crate::get_field;
use biliup_cli::server::common;
use biliup_cli::server::common::upload::submit_to_bilibili;
use biliup_cli::server::errors::{AppError, AppResult};
@@ -56,16 +59,39 @@ impl From<UploadLine> for biliup_cli::UploadLine {
}
}
#[derive(FromPyObject)]
/// `desc_v2` 的元素：`{"type", "raw_text", "biz_id"}` 形式的 dict，
/// 或带 `type_id` / `raw_text` / `biz_id` 属性的对象。`biz_id` 可省略。
pub struct PyCredit {
#[pyo3(item("type"))]
type_id: i8,
#[pyo3(item("raw_text"))]
raw_text: String,
#[pyo3(item("biz_id"))]
biz_id: Option<String>,
}
impl<'a, 'py> FromPyObject<'a, 'py> for PyCredit {
type Error = PyErr;
fn extract(obj: Borrowed<'a, 'py, PyAny>) -> PyResult<Self> {
let obj = obj.to_owned();
let type_key = if obj.cast::<PyMapping>().is_ok() {
"type"
} else {
"type_id"
};
let required = |key: &str| -> PyResult<Bound<'py, PyAny>> {
get_field(&obj, key)?
.ok_or_else(|| PyTypeError::new_err(format!("desc_v2 item is missing `{key}`")))
};
Ok(PyCredit {
type_id: required(type_key)?.extract()?,
raw_text: required("raw_text")?.extract()?,
biz_id: get_field(&obj, "biz_id")?
.map(|value| value.extract::<Option<String>>())
.transpose()?
.flatten(),
})
}
}
#[derive(Builder)]
pub struct StudioPre {
video_path: Vec<PathBuf>,
crates/stream-gears/tests/test_errors.py
+293
View File
@@ -0,0 +1,293 @@
"""Error semantics and argument types of the stream_gears Python API.
Run against an installed wheel: ``pytest crates/stream-gears/tests``. All tests
are offline: downloads hit a local ``http.server`` or a closed port, and
logins go through a proxy on a closed port.
"""
import http.server
import inspect
import socket
import threading
from types import SimpleNamespace
import pytest
import stream_gears
LOGIN_FUNCTIONS = [
"get_qrcode",
"login_by_cookies",
"login_by_qrcode",
"login_by_sms",
"login_by_web_cookies",
"login_by_web_qrcode",
"send_sms",
]
# FLV file header (9 bytes, audio + video) followed by PreviousTagSize0.
FLV_HEADER = b"FLV\x01\x05\x00\x00\x00\x09" + b"\x00\x00\x00\x00"
ROUTES = {
"/empty.flv": b"",
"/header-only.flv": FLV_HEADER,
# A tag header needs 11 bytes; the stream ends after 5.
"/truncated.flv": FLV_HEADER + b"\x09\x00\x00\x10\x00",
# A complete script tag whose 3-byte body is not valid AMF data.
"/bad-script.flv": FLV_HEADER
+ b"\x12\x00\x00\x03\x00\x00\x00\x00\x00\x00\x00"
+ b"\x02\x00\x00"
+ b"\x00\x00\x00\x0e",
"/not-a-stream": b"<html><body>hello</body></html>",
}
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
body = ROUTES.get(self.path)
if body is None:
self.send_error(404)
return
self.send_response(200)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
@pytest.fixture(scope="module")
def server():
httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
thread.start()
yield f"http://127.0.0.1:{httpd.server_address[1]}"
httpd.shutdown()
httpd.server_close()
@pytest.fixture
def closed_port():
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@pytest.fixture(autouse=True)
def in_tmp_path(tmp_path, monkeypatch):
# download() writes download.log and logins write cookies.json into the cwd.
monkeypatch.chdir(tmp_path)
def assert_stream_gears_error(call):
"""`call` must raise StreamGearsError, catchable with `except Exception`.
Catches BaseException so that a Rust panic (pyo3 PanicException derives
from BaseException) fails the assertion instead of aborting the run.
"""
try:
call()
except BaseException as e: # noqa: B036
assert isinstance(e, Exception), f"{type(e).__name__} is not an Exception: {e}"
assert type(e).__name__ == "StreamGearsError", f"{type(e).__name__}: {e}"
return e
pytest.fail("did not raise")
def download(url, tmp_path, segment=None, **kwargs):
if segment is None:
segment = stream_gears.PySegment()
return stream_gears.download_with_callback(
url, {}, str(tmp_path / "out"), segment, **kwargs
)
def test_stream_gears_error_hierarchy():
assert issubclass(stream_gears.StreamGearsError, RuntimeError)
assert issubclass(stream_gears.StreamGearsError, Exception)
@pytest.mark.parametrize("name", LOGIN_FUNCTIONS)
def test_login_proxy_defaults_to_none(name):
params = inspect.signature(getattr(stream_gears, name)).parameters
assert params["proxy"].default is None
@pytest.mark.parametrize("name", ["download", "download_with_callback"])
def test_download_signature(name):
params = inspect.signature(getattr(stream_gears, name)).parameters
assert list(params)[:4] == ["url", "header_map", "file_name", "segment"]
assert params["proxy"].default is None
def test_download_closed_port(tmp_path, closed_port):
assert_stream_gears_error(
lambda: download(f"http://127.0.0.1:{closed_port}/live.flv", tmp_path)
)
def test_download_http_error(server, tmp_path):
e = assert_stream_gears_error(lambda: download(f"{server}/missing", tmp_path))
assert "404" in str(e)
def test_download_empty_response(server, tmp_path):
e = assert_stream_gears_error(lambda: download(f"{server}/empty.flv", tmp_path))
assert "incomplete" in str(e)
def test_download_truncated_flv(server, tmp_path):
assert_stream_gears_error(lambda: download(f"{server}/truncated.flv", tmp_path))
def test_download_malformed_flv_tag(server, tmp_path):
e = assert_stream_gears_error(lambda: download(f"{server}/bad-script.flv", tmp_path))
assert "script tag" in str(e)
def test_download_hls_error(server, tmp_path):
e = assert_stream_gears_error(lambda: download(f"{server}/not-a-stream", tmp_path))
assert "HLS" in str(e)
def test_download_plain_function_rejects_bad_segment(server, tmp_path):
with pytest.raises(TypeError):
stream_gears.download(f"{server}/header-only.flv", {}, str(tmp_path / "out"), 42)
def test_callback_accepts_lambda(server, tmp_path):
names = []
download(f"{server}/header-only.flv", tmp_path, file_name_callback_fn=lambda n: names.append(n))
assert names == [str(tmp_path / "out.flv")]
assert (tmp_path / "out.flv").is_file()
def test_callback_accepts_bound_method(server, tmp_path):
names = []
download(f"{server}/header-only.flv", tmp_path, file_name_callback_fn=names.append)
assert names == [str(tmp_path / "out.flv")]
def test_callback_must_be_callable(server, tmp_path):
with pytest.raises(TypeError, match="callable"):
download(f"{server}/header-only.flv", tmp_path, file_name_callback_fn=42)
def test_callback_exception_is_reraised(server, tmp_path):
def callback(name):
raise ValueError(f"boom {name}")
with pytest.raises(ValueError, match="boom"):
download(f"{server}/header-only.flv", tmp_path, file_name_callback_fn=callback)
# The download itself finished before the callback error was re-raised.
assert (tmp_path / "out.flv").is_file()
def test_py_segment_fields_are_readable():
segment = stream_gears.PySegment()
assert segment.time is None and segment.size is None
segment.time = 60
segment.size = 10 * 1000 * 1000
assert (segment.time, segment.size) == (60, 10 * 1000 * 1000)
@pytest.mark.parametrize(
"segment",
[
{"time": 3600},
{"size": 1 << 30, "time": None},
{},
SimpleNamespace(time=3600, size=None),
SimpleNamespace(size=1 << 30),
],
ids=["dict-time", "dict-size", "dict-empty", "object", "object-size-only"],
)
def test_segment_accepts_dict_and_object(server, tmp_path, segment):
download(f"{server}/header-only.flv", tmp_path, segment=segment)
assert (tmp_path / "out.flv").is_file()
def upload(tmp_path, **kwargs):
kwargs.setdefault("cookie_file", str(tmp_path / "no-such-cookies.json"))
return stream_gears.upload([str(tmp_path / "video.flv")], title="t", **kwargs)
@pytest.mark.parametrize("extra_fields", ['{"a":1,}', "[1, 2]", "{"])
def test_upload_invalid_extra_fields(tmp_path, extra_fields):
with pytest.raises(ValueError, match="extra_fields"):
upload(tmp_path, extra_fields=extra_fields)
@pytest.mark.parametrize("extra_fields", ["", '{"a": 1}'])
def test_upload_valid_extra_fields_reach_login(tmp_path, extra_fields):
# Parsing passes; the missing cookie file is the next (offline) failure.
assert_stream_gears_error(lambda: upload(tmp_path, extra_fields=extra_fields))
@pytest.mark.parametrize(
"credit",
[
{"type": 1, "raw_text": "hi", "biz_id": ""},
{"type": 1, "raw_text": "hi"},
SimpleNamespace(type_id=1, raw_text="hi", biz_id=None),
],
ids=["dict", "dict-no-biz-id", "object"],
)
def test_upload_desc_v2_accepts_dict_and_object(tmp_path, credit):
assert_stream_gears_error(lambda: upload(tmp_path, desc_v2=[credit]))
def test_upload_desc_v2_missing_field(tmp_path):
with pytest.raises(TypeError, match="raw_text"):
upload(tmp_path, desc_v2=[{"type": 1}])
@pytest.mark.parametrize("name", ["login_by_sms", "login_by_qrcode"])
def test_login_invalid_json_raises_value_error(name):
args = (123456, "{not json") if name == "login_by_sms" else ("{not json",)
with pytest.raises(ValueError, match="ret"):
getattr(stream_gears, name)(*args, proxy=None)
def test_login_by_sms_failure_raises(tmp_path, closed_port):
e = assert_stream_gears_error(
lambda: stream_gears.login_by_sms(
123456,
'{"captcha_key": "x", "tel": "13800000000", "cid": 86}',
proxy=f"http://127.0.0.1:{closed_port}",
)
)
assert str(e)
assert not (tmp_path / "cookies.json").exists()
def test_login_by_cookies_missing_file(tmp_path):
e = assert_stream_gears_error(
lambda: stream_gears.login_by_cookies(str(tmp_path / "no-such-cookies.json"))
)
assert "no-such-cookies.json" in str(e) or "No such file" in str(e)
@pytest.mark.parametrize("name", ["login_by_sms", "login_by_web_cookies", "login_by_web_qrcode"])
def test_login_file_defaults_to_cookies_json(name):
params = inspect.signature(getattr(stream_gears, name)).parameters
assert params["file"].default == "cookies.json"
def test_login_by_qrcode_file_is_optional():
params = inspect.signature(stream_gears.login_by_qrcode).parameters
assert params["file"].default is None
def test_login_by_sms_failure_does_not_write_file(tmp_path, closed_port):
target = tmp_path / "sub" / "my-cookies.json"
assert_stream_gears_error(
lambda: stream_gears.login_by_sms(
123456,
'{"captcha_key": "x", "tel": "13800000000", "cid": 86}',
proxy=f"http://127.0.0.1:{closed_port}",
file=target,
)
)
assert not target.exists()
assert not (tmp_path / "cookies.json").exists()
crates/stream-gears/tests/test_smoke.py
+1 -23
View File
@@ -5,7 +5,6 @@ source tree: ``pytest crates/stream-gears/tests``. All tests are offline.
"""
import importlib.metadata
import inspect
import subprocess
import sys
import sysconfig
@@ -16,6 +15,7 @@ import stream_gears
EXPORTS = {
"PySegment",
"StreamGearsError",
"UploadLine",
"config_bindings",
"download",
@@ -51,16 +51,6 @@ UPLOAD_LINES = [
"Akbd",
]
LOGIN_FUNCTIONS = [
"get_qrcode",
"login_by_cookies",
"login_by_qrcode",
"login_by_sms",
"login_by_web_cookies",
"login_by_web_qrcode",
"send_sms",
]
# The CLI tests go through the installed console script / `python -m biliup`,
# so they run in a subprocess. cwd is a temp dir so the repo's `biliup/` source
# package can't shadow the installed one.
@@ -96,7 +86,6 @@ def test_upload_line_members():
def test_py_segment():
# Fields are write-only (`#[pyclass(set_all)]` without `get_all`).
segment = stream_gears.PySegment()
segment.time = 60
segment.size = 10 * 1000 * 1000
@@ -112,17 +101,6 @@ def test_config_bindings():
config.get("no_such_key")
@pytest.mark.parametrize("name", LOGIN_FUNCTIONS)
@pytest.mark.xfail(
strict=True,
raises=AssertionError,
reason="PyO3 >= 0.24 no longer defaults trailing Option args to None; login functions need proxy=None",
)
def test_login_proxy_is_optional(name):
params = inspect.signature(getattr(stream_gears, name)).parameters
assert params["proxy"].default is None
def test_biliup_version(tmp_path):
result = run([biliup_script(), "--version"], tmp_path)
assert result.returncode == 0, result.stderr