diff --git a/crates/biliup/src/downloader/httpflv.rs b/crates/biliup/src/downloader/httpflv.rs index 0bbe61f..6c2cbb8 100644 --- a/crates/biliup/src/downloader/httpflv.rs +++ b/crates/biliup/src/downloader/httpflv.rs @@ -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, diff --git a/crates/stream-gears/src/lib.rs b/crates/stream-gears/src/lib.rs index 698a783..9abccc5 100644 --- a/crates/stream-gears/src/lib.rs +++ b/crates/stream-gears/src/lib.rs @@ -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::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::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>> { + if let Ok(mapping) = obj.cast::() { + 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, - // #[pyo3(attribute("size"))] size: Option, } @@ -85,6 +136,44 @@ impl PySegment { } } +/// `download` 的 `segment` 参数：`PySegment`、带 `time` / `size` 键的 dict， +/// 或带 `time` / `size` 属性的对象。 +struct SegmentArg { + time: Option, + size: Option, +} + +impl<'a, 'py> FromPyObject<'a, 'py> for SegmentArg { + type Error = PyErr; + + fn extract(obj: Borrowed<'a, 'py, PyAny>) -> PyResult { + if let Ok(segment) = obj.cast::() { + let segment = segment.borrow(); + return Ok(SegmentArg { + time: segment.time, + size: segment.size, + }); + } + let obj = obj.to_owned(); + if obj.cast::().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> { + Ok(get_field(&obj, key)? + .map(|value| value.extract::>()) + .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, file_name: &str, - segment: PySegment, + segment: SegmentArg, proxy: Option, ) -> 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, file_name: &str, - segment: PySegment, - file_name_callback_fn: Option>, + segment: SegmentArg, + file_name_callback_fn: Option>, proxy: Option, ) -> 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>> = 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::().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) -> PyResult { 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, ) -> PyResult { 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) -> PyResult { +#[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, + file: PathBuf, +) -> PyResult { + 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) -> PyResult { 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) -> PyResult { +#[pyo3(signature = (ret, proxy=None, file=None))] +fn login_by_qrcode( + py: Python<'_>, + ret: String, + proxy: Option, + file: Option, +) -> PyResult { + 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, + file: PathBuf, ) -> PyResult { 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, + file: PathBuf, ) -> PyResult { 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, proxy: Option, ) -> 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::()?; m.add_class::()?; + m.add("StreamGearsError", m.py().get_type::())?; Ok(()) } -fn parse_extra_fields(s: Option) -> HashMap { - 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> { + 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}"); + } } } diff --git a/crates/stream-gears/src/login.rs b/crates/stream-gears/src/login.rs index 5c8aeca..1bb51f2 100644 --- a/crates/stream-gears/src/login.rs +++ b/crates/stream-gears/src/login.rs @@ -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 { 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 { 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 { 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 { 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) diff --git a/crates/stream-gears/src/uploader.rs b/crates/stream-gears/src/uploader.rs index 91ed575..0f5830e 100644 --- a/crates/stream-gears/src/uploader.rs +++ b/crates/stream-gears/src/uploader.rs @@ -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 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, } +impl<'a, 'py> FromPyObject<'a, 'py> for PyCredit { + type Error = PyErr; + + fn extract(obj: Borrowed<'a, 'py, PyAny>) -> PyResult { + let obj = obj.to_owned(); + let type_key = if obj.cast::().is_ok() { + "type" + } else { + "type_id" + }; + let required = |key: &str| -> PyResult> { + 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::>()) + .transpose()? + .flatten(), + }) + } +} + #[derive(Builder)] pub struct StudioPre { video_path: Vec, diff --git a/crates/stream-gears/tests/test_errors.py b/crates/stream-gears/tests/test_errors.py new file mode 100644 index 0000000..4f17a52 --- /dev/null +++ b/crates/stream-gears/tests/test_errors.py @@ -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"hello", +} + + +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() diff --git a/crates/stream-gears/tests/test_smoke.py b/crates/stream-gears/tests/test_smoke.py index 7158be7..e939427 100644 --- a/crates/stream-gears/tests/test_smoke.py +++ b/crates/stream-gears/tests/test_smoke.py @@ -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