mocra_core/common/model/
request.rs

1use crate::cacheable::CacheAble;
2use crate::common::interface::storage::{BlobStorage, Offloadable};
3use crate::common::model::ExecutionMark;
4use crate::common::model::ModuleConfig;
5use crate::common::model::login_info::LoginInfo;
6use crate::common::model::meta::MetaData;
7use crate::common::model::{Cookies, Headers};
8use crate::utils::encrypt::md5;
9use async_trait::async_trait;
10use log::warn;
11use mocra_proxy::ProxyEnum;
12use once_cell::sync::OnceCell;
13use serde::{Deserialize, Serialize};
14use std::fmt::Display;
15use std::sync::Arc;
16use uuid::Uuid;
17
18pub enum RequestMethod {
19    Post,
20    Get,
21    Delete,
22    Options,
23    Put,
24    Head,
25    // `wss` is currently just a marker used to distinguish WebSocket requests.
26    // Actual WebSocket calls do not require this parameter.
27    Wss,
28}
29impl Display for RequestMethod {
30    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
31        let str = match self {
32            RequestMethod::Post => "POST".to_string(),
33            RequestMethod::Get => "GET".to_string(),
34            RequestMethod::Delete => "DELETE".to_string(),
35            RequestMethod::Options => "OPTIONS".to_string(),
36            RequestMethod::Put => "PUT".to_string(),
37            RequestMethod::Head => "HEAD".to_string(),
38            RequestMethod::Wss => "WSS".to_string(),
39        };
40        write!(f, "{str}")
41    }
42}
43
44impl AsRef<str> for RequestMethod {
45    fn as_ref(&self) -> &str {
46        match self {
47            RequestMethod::Post => "POST",
48            RequestMethod::Get => "GET",
49            RequestMethod::Delete => "DELETE",
50            RequestMethod::Options => "OPTIONS",
51            RequestMethod::Put => "PUT",
52            RequestMethod::Head => "HEAD",
53            RequestMethod::Wss => "WSS",
54        }
55    }
56}
57impl From<RequestMethod> for String {
58    fn from(val: RequestMethod) -> Self {
59        val.to_string()
60    }
61}
62
63#[derive(Debug, Clone, Serialize, Deserialize)]
64pub struct Request {
65    pub id: Uuid,
66    pub platform: String,
67    pub account: String,
68    pub module: String,
69    pub url: String,
70    pub method: String,
71    pub headers: Headers,
72    pub cookies: Cookies,
73    pub retry_times: usize,
74    pub task_retry_times: usize,
75    pub use_new_client: bool,
76    pub timeout: u64,
77    /// `meta` from `ParserModel` and `ErrorModel` is appended to `Task.meta` in `TaskFactory`.
78    ///
79    /// `meta` stores extra information. In `Module.generate`, `Task.meta`,
80    /// `ShopInfo.extra`, and `ModuleConfig` are merged into `Request.meta`.
81    /// Metadata sources include `Task.meta`, `LoginInfo.extra`, `ModuleConfig`, and trait-level
82    /// additions. Caller-defined fields are used to distinguish task/login/trait data.
83    pub meta: MetaData,
84    pub params: Option<Vec<(String, String)>>,
85    #[serde(with = "crate::common::model::serde_value::option_value")]
86    pub json: Option<serde_json::Value>,
87    pub body: Option<Vec<u8>>,
88    #[serde(with = "crate::common::model::serde_value::option_value")]
89    pub form: Option<serde_json::Value>,
90    /// Header names that should be cached.
91    /// Example: `Cache-Control`, `Expires`, `ETag`.
92    pub cache_headers: Option<Vec<String>>,
93    pub proxy: Option<ProxyEnum>,
94    /// Rate-limit identifier.
95    /// Used to mark requests that belong to the same rate-limit group.
96    /// Defaults to `module_id`.
97    pub limit_id: String,
98    pub download_middleware: Vec<String>,
99    pub data_middleware: Vec<String>,
100    pub task_finished: bool,
101    pub time_sleep_secs: Option<u64>,
102    pub context: ExecutionMark,
103    pub run_id: Uuid,
104    pub prefix_request: Uuid,
105    /// Custom hash string used to override default request hash calculation.
106    pub hash_str: Option<String>,
107    #[serde(default)]
108    pub enable_session: bool,
109    #[serde(default)]
110    pub enable_response_cache: bool,
111    /// Enable distributed lock for this request to ensure serial execution within the same task/run
112    pub enable_locker: Option<bool>,
113    pub downloader: String,
114    #[serde(default)]
115    pub priority: crate::common::model::Priority,
116    #[serde(skip)]
117    hash_cache: OnceCell<String>,
118}
119
120#[async_trait]
121impl Offloadable for Request {
122    fn should_offload(&self, _threshold: usize) -> bool {
123        false
124    }
125    async fn offload(&mut self, _storage: &Arc<dyn BlobStorage>) -> crate::errors::Result<()> {
126        Ok(())
127    }
128    async fn reload(&mut self, _storage: &Arc<dyn BlobStorage>) -> crate::errors::Result<()> {
129        Ok(())
130    }
131}
132
133impl Request {
134    pub fn new(url: impl AsRef<str>, method: impl AsRef<str>) -> Self {
135        Request {
136            id: Uuid::now_v7(),
137            platform: "".to_string(),
138            account: "".to_string(),
139            module: "".to_string(),
140            url: url.as_ref().into(),
141            method: method.as_ref().into(),
142            headers: Headers::default(),
143            cookies: Cookies::default(),
144            retry_times: 0,
145            meta: MetaData::default(),
146            params: None,
147            json: None,
148            body: None,
149            form: None,
150            timeout: 30, // Default timeout: 30 seconds.
151            cache_headers: None,
152            proxy: None,
153            limit_id: "".to_string(),
154            use_new_client: false,
155            download_middleware: vec![],
156            data_middleware: vec![],
157            task_retry_times: 0,
158            task_finished: false,
159            time_sleep_secs: None,
160            context: Default::default(),
161            run_id: Uuid::now_v7(),
162            prefix_request: Uuid::nil(),
163            hash_str: None,
164            enable_session: false,
165            enable_response_cache: false,
166            enable_locker: None,
167            downloader: "request_downloader".to_string(),
168            priority: crate::common::model::Priority::default(),
169            hash_cache: OnceCell::new(),
170        }
171    }
172    pub fn with_priority(mut self, priority: crate::common::model::Priority) -> Self {
173        self.priority = priority;
174        self
175    }
176    pub fn use_proxy(&mut self, proxy: ProxyEnum) -> &mut Request {
177        self.proxy = Some(proxy);
178        self
179    }
180    pub fn task_id(&self) -> String {
181        format!("{}-{}", self.account, self.platform)
182    }
183    pub fn module_id(&self) -> String {
184        format!("{}-{}-{}", self.account, self.platform, self.module)
185    }
186    /// Run-scoped task identifier for error tracking. Includes `run_id` to isolate
187    /// error state across different DAG runs.
188    pub fn task_runtime_id(&self) -> String {
189        format!("{}:{}:{}", self.platform, self.account, self.run_id)
190    }
191    /// Run-scoped module identifier for error tracking. Includes `run_id` to isolate
192    /// error state across different DAG runs.
193    pub fn module_runtime_id(&self) -> String {
194        format!(
195            "{}-{}-{}-{}",
196            self.account, self.platform, self.module, self.run_id
197        )
198    }
199    pub fn with_params(mut self, params: Vec<(impl AsRef<str>, impl AsRef<str>)>) -> Self {
200        self.params = Some(
201            params
202                .iter()
203                .map(|(a, b)| (a.as_ref().to_string(), b.as_ref().to_string()))
204                .collect(),
205        );
206        self
207    }
208    pub fn with_headers(mut self, headers: Headers) -> Self {
209        self.headers.merge(&headers);
210        self
211    }
212    pub fn with_cookies(mut self, cookies: Cookies) -> Self {
213        self.cookies.merge(&cookies);
214        self
215    }
216    pub fn with_json<T: Serialize + ?Sized>(mut self, json: &T) -> Self {
217        match serde_json::to_value(json) {
218            Ok(value) => {
219                self.json = Some(value);
220            }
221            Err(e) => {
222                warn!("Request::with_json serialization failed: {}", e);
223            }
224        }
225        self
226    }
227    pub fn with_body(mut self, body: Vec<u8>) -> Self {
228        self.body = Some(body);
229        self
230    }
231    pub fn with_form<T: Serialize + ?Sized>(mut self, form: &T) -> Self {
232        match serde_json::to_value(form) {
233            Ok(value) => {
234                self.form = Some(value);
235            }
236            Err(e) => {
237                warn!("Request::with_form serialization failed: {}", e);
238            }
239        }
240        self
241    }
242    pub fn with_meta<T>(mut self, meta: T) -> Self
243    where
244        T: Serialize,
245    {
246        if let Ok(serde_json::Value::Object(map)) = serde_json::to_value(meta) {
247            for (key, value) in map {
248                self.meta = self.meta.add_trait_config(key, value);
249            }
250        }
251        self
252    }
253    pub fn add_meta<T>(mut self, key: impl AsRef<str>, value: T) -> Self
254    where
255        T: Serialize,
256    {
257        self.meta = self.meta.add_trait_config(key, value);
258        self
259    }
260    pub fn with_login_info(mut self, info: &LoginInfo) -> Self {
261        self.meta = self.meta.add_login_info(info);
262        self
263    }
264    pub fn with_task_config<T>(mut self, task_meta: T) -> Self
265    where
266        T: Serialize + for<'de> Deserialize<'de>,
267    {
268        self.meta = self.meta.add_task_config(task_meta);
269        self
270    }
271    pub fn with_module_config(mut self, value: &ModuleConfig) -> Self {
272        self.meta = self.meta.add_module_config(value);
273        self
274    }
275    pub fn with_sleep(mut self, secs: u64) -> Self {
276        self.time_sleep_secs = Some(secs);
277        self
278    }
279    // Typed Context helpers
280    pub fn with_context(mut self, ctx: ExecutionMark) -> Self {
281        self.context = ctx;
282        self
283    }
284    pub fn hash(&self) -> String {
285        // Build a canonical representation of the request's identity-related fields
286        if let Some(hash) = &self.hash_str {
287            return hash.to_owned();
288        }
289        if let Some(cached) = self.hash_cache.get() {
290            return cached.clone();
291        }
292        let canonical = format!(
293            "{},{},{},{},{},{},{},{:?},{},{}",
294            self.account,
295            self.platform,
296            self.module,
297            self.url,
298            self.method,
299            serde_json::to_string(&self.params).unwrap_or_default(),
300            self.json.as_ref().unwrap_or_default(),
301            self.body.as_deref().unwrap_or(&[]),
302            self.form.as_ref().unwrap_or_default(),
303            self.run_id
304        );
305
306        // MD5 hash in lowercase hex for stable, compact identity
307        let digest = md5(canonical.as_bytes()).to_string();
308        let _ = self.hash_cache.set(digest.clone());
309        digest
310    }
311    pub fn enable_session(mut self, enable: bool) -> Self {
312        self.enable_session = enable;
313        self
314    }
315    pub fn enable_response_cache(mut self, enable: bool) -> Self {
316        self.enable_response_cache = enable;
317        self
318    }
319    pub fn enable_response_cache_with<T>(mut self, hash_able: &T) -> Self
320    where
321        T: Serialize,
322    {
323        if let Ok(hash_str) = serde_json::to_string(hash_able) {
324            self.enable_response_cache = true;
325            self.hash_str = Some(md5(hash_str.as_bytes()));
326        }
327        self
328    }
329}
330
331impl CacheAble for Request {
332    fn field() -> impl AsRef<str> {
333        "request"
334    }
335
336    fn serialized_size_hint(&self) -> Option<usize> {
337        Some(
338            self.url.len()
339                + self.method.len()
340                + self.headers.headers.len() * 64
341                + self.cookies.cookies.len() * 64
342                + self.body.as_ref().map_or(0, |b| b.len())
343                + self.form.as_ref().map_or(0, |_| 512)
344                + self.json.as_ref().map_or(0, |_| 512),
345        )
346    }
347
348    fn clone_for_serialize(&self) -> Option<Self> {
349        Some(self.clone())
350    }
351}
352
353impl crate::common::model::priority::Prioritizable for Request {
354    fn get_priority(&self) -> crate::common::model::Priority {
355        self.priority
356    }
357}
mocra_core/common/model/request.rs

mocra_core/common/model/
request.rs
